apunkt/plano
1
0
Fork 0
mirror of https://github.com/katanemo/plano.git synced 2026-05-09 15:52:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

updated crew

Browse source
This commit is contained in:
Ahmed Burney 2026-01-28 21:11:23 +05:00
parent b6702c1c49
commit fb04cbb2ce
1 changed files with 313 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

339
demos/use_cases/credit_risk_case_copilot/src/credit_risk_demo/risk_crew_agent.py
View file

@ -77,11 +77,9 @@ class RiskAssessmentResult(BaseModel):
human_response: str
def create_risk_crew(application_data: Dict[str, Any]) -> Crew:
"""Create a CrewAI crew for risk assessment with 4 specialized agents."""
# Agent 1: Intake & Normalization Specialist
intake_agent = Agent(
def build_intake_agent() -> Agent:
"""Build the intake & normalization agent."""
return Agent(
role="Loan Intake & Normalization Specialist",
goal="Extract, validate, and normalize loan application data for downstream risk assessment",
backstory="""You are an expert at processing loan applications from various sources.
@ -93,8 +91,10 @@ def create_risk_crew(application_data: Dict[str, Any]) -> Crew:
allow_delegation=False,
)
# Agent 2: Risk Scoring & Driver Analysis Expert
risk_scoring_agent = Agent(
def build_risk_agent() -> Agent:
"""Build the risk scoring & driver analysis agent."""
return Agent(
role="Risk Scoring & Driver Analysis Expert",
goal="Calculate comprehensive risk scores and identify key risk drivers with evidence",
backstory="""You are a senior credit risk analyst with 15+ years experience. You analyze:
@ -111,8 +111,10 @@ def create_risk_crew(application_data: Dict[str, Any]) -> Crew:
allow_delegation=False,
)
# Agent 3: Policy & Compliance Officer
policy_agent = Agent(
def build_policy_agent() -> Agent:
"""Build the policy & compliance agent."""
return Agent(
role="Policy & Compliance Officer",
goal="Verify compliance with lending policies and identify exceptions",
backstory="""You are a compliance expert ensuring all loan applications meet regulatory
@ -129,8 +131,10 @@ def create_risk_crew(application_data: Dict[str, Any]) -> Crew:
allow_delegation=False,
)
# Agent 4: Decision Memo & Action Specialist
memo_agent = Agent(
def build_memo_agent() -> Agent:
"""Build the decision memo & action agent."""
return Agent(
role="Decision Memo & Action Specialist",
goal="Generate bank-ready decision memos and recommend clear actions",
backstory="""You are a senior credit officer who writes clear, concise decision memos
@ -147,6 +151,184 @@ def create_risk_crew(application_data: Dict[str, Any]) -> Crew:
allow_delegation=False,
)
def make_intake_task(application_data: Dict[str, Any], agent: Agent) -> Task:
"""Build the intake task prompt."""
return Task(
description=f"""Analyze this loan application and extract all relevant information:
{json.dumps(application_data, indent=2)}
Extract and normalize:
1. Applicant name and loan amount
2. Monthly income and employment status
3. Credit score and existing loans
4. Total debt and delinquencies
5. Credit utilization rate
6. KYC, income verification, and address verification status
7. Calculate DTI if income is available: (total_debt / monthly_income) * 100
8. Flag any missing critical fields
Output JSON only with:
- step: "intake"
- normalized_data: object of normalized fields
- missing_fields: list of missing critical fields""",
agent=agent,
expected_output="JSON only with normalized data and missing fields",
)
def make_risk_task(payload: Dict[str, Any], agent: Agent) -> Task:
"""Build the risk scoring task prompt."""
return Task(
description=f"""You are given an input payload that includes the application and intake output:
{json.dumps(payload, indent=2)}
Use intake.normalized_data for your analysis.
**Risk Scoring Criteria:**
1. **Credit Score Assessment:**
- Excellent (≥750): Low risk
- Good (650-749): Medium risk
- Fair (550-649): High risk
- Poor (<550): Critical risk
- Missing: Medium risk (thin file)
2. **Debt-to-Income Ratio:**
- <35%: Low risk
- 35-50%: Medium risk
- >50%: Critical risk
- Missing: High risk
3. **Delinquency History:**
- 0: Low risk
- 1-2: Medium risk
- >2: Critical risk
4. **Credit Utilization:**
- <30%: Low risk
- 30-70%: Medium risk
- >70%: High risk
Output JSON only with:
- step: "risk"
- risk_band: LOW|MEDIUM|HIGH
- confidence_score: 0.0-1.0
- top_3_risk_drivers: [{{
"factor": string,
"impact": CRITICAL|HIGH|MEDIUM|LOW,
"evidence": string
}}]""",
agent=agent,
expected_output="JSON only with risk band, confidence, and top drivers",
)
def make_policy_task(payload: Dict[str, Any], agent: Agent) -> Task:
"""Build the policy compliance task prompt."""
return Task(
description=f"""You are given an input payload that includes the application, intake, and risk output:
{json.dumps(payload, indent=2)}
Use intake.normalized_data and risk outputs.
**Policy Checks:**
1. KYC Completion: Check if CNIC, phone, and address are provided
2. Income Verification: Check if income is verified
3. Address Verification: Check if address is verified
4. DTI Limit: Flag if DTI >60% (automatic reject threshold)
5. Credit Score: Flag if <500 (minimum acceptable)
6. Delinquencies: Flag if >2 in recent history
**Required Documents by Risk Band:**
- LOW: Valid CNIC, Credit Report, Employment Letter, Bank Statements (3 months)
- MEDIUM: + Income proof (6 months), Address proof, Tax Returns (2 years)
- HIGH: + Guarantor Documents, Collateral Valuation, Detailed Financials
Output JSON only with:
- step: "policy"
- policy_checks: [{{"check": string, "status": PASS|FAIL|WARNING, "details": string}}]
- exceptions: [string]
- required_documents: [string]""",
agent=agent,
expected_output="JSON only with policy checks, exceptions, and required documents",
)
def make_memo_task(payload: Dict[str, Any], agent: Agent) -> Task:
"""Build the decision memo task prompt."""
return Task(
description=f"""You are given an input payload that includes the application, intake, risk, and policy output:
{json.dumps(payload, indent=2)}
Generate a concise memo and recommendation.
**Recommendation Rules:**
- APPROVE: LOW risk + all checks passed
- CONDITIONAL_APPROVE: LOW/MEDIUM risk + minor issues (collect docs)
- REFER: MEDIUM/HIGH risk + exceptions (manual review)
- REJECT: HIGH risk OR critical policy violations (>60% DTI, <500 credit score)
Output JSON only with:
- step: "memo"
- recommended_action: APPROVE|CONDITIONAL_APPROVE|REFER|REJECT
- decision_memo: string (max 300 words)""",
agent=agent,
expected_output="JSON only with recommended action and decision memo",
)
def run_single_step(agent: Agent, task: Task) -> str:
"""Run a single-step CrewAI workflow and return the output."""
crew = Crew(
agents=[agent],
tasks=[task],
process=Process.sequential,
verbose=True,
)
return crew.kickoff()
def extract_json_from_content(content: str) -> Optional[Dict[str, Any]]:
"""Extract a JSON object from a message content string."""
try:
if "{" in content and "}" in content:
json_start = content.index("{")
json_end = content.rindex("}") + 1
json_str = content[json_start:json_end]
return json.loads(json_str)
except Exception as e:
logger.warning(f"Could not parse JSON from message: {e}")
return None
def extract_json_block(output_text: str) -> Optional[Dict[str, Any]]:
"""Extract the first JSON object or fenced JSON block from output text."""
try:
if "```json" in output_text:
json_start = output_text.index("```json") + 7
json_end = output_text.index("```", json_start)
return json.loads(output_text[json_start:json_end].strip())
if "{" in output_text and "}" in output_text:
json_start = output_text.index("{")
json_end = output_text.rindex("}") + 1
return json.loads(output_text[json_start:json_end])
except Exception as e:
logger.warning(f"Could not parse JSON from output: {e}")
return None
def create_risk_crew(application_data: Dict[str, Any]) -> Crew:
"""Create a CrewAI crew for risk assessment with 4 specialized agents."""
intake_agent = build_intake_agent()
risk_scoring_agent = build_risk_agent()
policy_agent = build_policy_agent()
memo_agent = build_memo_agent()
# Task 1: Intake and normalization
intake_task = Task(
description=f"""Analyze this loan application and extract all relevant information:
@ -554,21 +736,8 @@ async def chat_completions(request: Request):
logger.info(f"Processing CrewAI request {request_id}: {content[:100]}")
# Try to parse JSON from content
application_data = {}
try:
if "{" in content and "}" in content:
json_start = content.index("{")
json_end = content.rindex("}") + 1
json_str = content[json_start:json_end]
application_data = json.loads(json_str)
else:
# Simple request without JSON
application_data = {
"applicant_name": "Sample Applicant",
"loan_amount": 100000,
}
except Exception as e:
logger.warning(f"Could not parse JSON from message: {e}")
application_data = extract_json_from_content(content)
if not application_data:
application_data = {
"applicant_name": "Sample Applicant",
"loan_amount": 100000,
@ -628,6 +797,124 @@ async def chat_completions(request: Request):
)
async def handle_single_agent_step(request: Request, step: str) -> JSONResponse:
"""Handle a single-step agent request with OpenAI-compatible response."""
with tracer.start_as_current_span(f"{step}_chat_completions") as span:
try:
body = await request.json()
messages = body.get("messages", [])
request_id = str(uuid.uuid4())
span.set_attribute("request_id", request_id)
span.set_attribute("step", step)
last_user_msg = next(
(m for m in reversed(messages) if m.get("role") == "user"), None
)
if not last_user_msg:
return JSONResponse(
status_code=400, content={"error": "No user message found"}
)
content = last_user_msg.get("content", "")
logger.info(f"Processing {step} request {request_id}: {content[:100]}")
payload = extract_json_from_content(content)
if not payload:
return JSONResponse(
status_code=400,
content={"error": "No JSON payload found in user message"},
)
if step == "intake":
agent = build_intake_agent()
task = make_intake_task(payload, agent)
model_name = "loan_intake_agent"
human_response = "Intake normalization complete."
elif step == "risk":
agent = build_risk_agent()
task = make_risk_task(payload, agent)
model_name = "risk_scoring_agent"
human_response = "Risk scoring complete."
elif step == "policy":
agent = build_policy_agent()
task = make_policy_task(payload, agent)
model_name = "policy_compliance_agent"
human_response = "Policy compliance review complete."
elif step == "memo":
agent = build_memo_agent()
task = make_memo_task(payload, agent)
model_name = "decision_memo_agent"
human_response = "Decision memo complete."
else:
return JSONResponse(
status_code=400, content={"error": f"Unknown step: {step}"}
)
crew_output = run_single_step(agent, task)
json_payload = extract_json_block(str(crew_output)) or {"step": step}
response_content = (
f"{human_response}\n\n```json\n{json.dumps(json_payload, indent=2)}\n```"
)
return JSONResponse(
content={
"id": f"chatcmpl-{request_id}",
"object": "chat.completion",
"created": int(datetime.utcnow().timestamp()),
"model": model_name,
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": response_content,
},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
},
"metadata": {
"framework": "CrewAI",
"step": step,
"request_id": request_id,
},
}
)
except Exception as e:
logger.error(f"Error processing {step} request: {e}", exc_info=True)
span.record_exception(e)
return JSONResponse(
status_code=500, content={"error": str(e), "framework": "CrewAI"}
)
@app.post("/v1/agents/intake/chat/completions")
async def intake_chat_completions(request: Request):
return await handle_single_agent_step(request, "intake")
@app.post("/v1/agents/risk/chat/completions")
async def risk_chat_completions(request: Request):
return await handle_single_agent_step(request, "risk")
@app.post("/v1/agents/policy/chat/completions")
async def policy_chat_completions(request: Request):
return await handle_single_agent_step(request, "policy")
@app.post("/v1/agents/memo/chat/completions")
async def memo_chat_completions(request: Request):
return await handle_single_agent_step(request, "memo")
@app.get("/health")
async def health_check():
"""Health check endpoint."""