第九篇 Agent 架构设计#

用户提交理赔
  → Triage Agent (分流)
  → Claim Processor (初审)
  → 风控 Agent / 法务 Agent (专业处理)
  → 结案

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class InsuranceState(TypedDict):
    """全局共享状态"""
    messages: Annotated[list, add_messages]  # 对话历史（自动合并）
    claim_id: str                             # 案件ID
    risk_score: float                         # 风险评分
    current_agent: str                        # 当前处理人

from langchain_core.tools import tool

def create_transfer_tool(target_agent: str):
    """工厂函数：生成移交工具"""
    @tool(f"transfer_to_{target_agent}")
    def transfer() -> str:
        f"""将任务移交给 {target_agent} 继续处理。"""
        return target_agent
    return transfer

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# === Triage Agent ===
triage_tools = [
    create_transfer_tool("ClaimProcessor")
]

def triage_node(state: InsuranceState):
    """分流专员：收集信息并转发"""
    system_msg = SystemMessage(content="""
    你是理赔分流专员。职责：
    1. 询问用户案件编号和基本情况
    2. 记录到系统后，调用 transfer_to_ClaimProcessor 移交
    """)

    response = llm.bind_tools(triage_tools).invoke(
        [system_msg] + state["messages"]
    )
    return {"messages": [response]}

# === Claim Processor ===
processor_tools = [
    create_transfer_tool("FraudDetector"),
    create_transfer_tool("LegalAdvisor")
]

def processor_node(state: InsuranceState):
    """初审专员：计算风险并分流"""
    system_msg = SystemMessage(content="""
    你是理赔初审员。职责：
    1. 分析案件，计算风险评分（0-100）
    2. 如果风险 > 70，调用 transfer_to_FraudDetector
    3. 否则调用 transfer_to_LegalAdvisor
    """)

    response = llm.bind_tools(processor_tools).invoke(
        [system_msg] + state["messages"]
    )
    return {"messages": [response]}

# === Fraud Detector ===
def fraud_node(state: InsuranceState):
    """风控专员：终端节点，不再移交"""
    system_msg = SystemMessage(content="你是风控专员，进行最终审核。")
    response = llm.invoke([system_msg] + state["messages"])
    return {"messages": [response]}

# === Legal Advisor ===
def legal_node(state: InsuranceState):
    """法务专员：终端节点"""
    system_msg = SystemMessage(content="你是法务顾问，提供法律意见。")
    response = llm.invoke([system_msg] + state["messages"])
    return {"messages": [response]}

from langgraph.graph import StateGraph, END

# 1. 初始化图
workflow = StateGraph(InsuranceState)

# 2. 添加节点
workflow.add_node("Triage", triage_node)
workflow.add_node("ClaimProcessor", processor_node)
workflow.add_node("FraudDetector", fraud_node)
workflow.add_node("LegalAdvisor", legal_node)

# 3. 定义路由函数
def router(state: InsuranceState):
    """根据最后一条消息的 Tool Call 决定路由"""
    last_msg = state["messages"][-1]

    # 检查是否有工具调用
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        tool_call = last_msg.tool_calls[0]
        tool_name = tool_call["name"]

        # 如果是 transfer_to_X 工具，返回目标节点名
        if tool_name.startswith("transfer_to_"):
            return tool_name.replace("transfer_to_", "")

    # 否则结束
    return END

# 4. 添加条件边（核心机制）
workflow.add_conditional_edges("Triage", router)
workflow.add_conditional_edges("ClaimProcessor", router)

# 5. 终端节点直接结束
workflow.add_edge("FraudDetector", END)
workflow.add_edge("LegalAdvisor", END)

# 6. 设置入口
workflow.set_entry_point("Triage")

# 7. 编译
app = workflow.compile()

from langchain_core.messages import HumanMessage

result = app.invoke({
    "messages": [HumanMessage(content="我要报案，案件号 C12345")],
    "claim_id": "",
    "risk_score": 0.0,
    "current_agent": ""
})

print(result["messages"][-1].content)

Triage: 收集信息 → 调用 transfer_to_ClaimProcessor
  ↓
ClaimProcessor: 计算风险 → 调用 transfer_to_FraudDetector
  ↓
FraudDetector: 最终审核 → END

pip install langgraph-checkpoint-redis redis

from langgraph.checkpoint.redis import RedisSaver
import redis

# 1. 连接 Redis
redis_client = redis.Redis(host="localhost", port=6379, db=0)

# 2. 创建 Checkpointer
checkpointer = RedisSaver(redis_client)

# 3. 编译时传入
app = workflow.compile(checkpointer=checkpointer)

# worker_triage.py
from fastapi import FastAPI
from langserve import add_routes

# 只包含 Triage 节点的子图
triage_graph = create_triage_subgraph()
triage_app = triage_graph.compile(checkpointer=checkpointer)

api = FastAPI()
add_routes(api, triage_app, path="/triage")

# 启动：uvicorn worker_triage:api --port 8001

# worker_fraud.py
fraud_graph = create_fraud_subgraph()
fraud_app = fraud_graph.compile(checkpointer=checkpointer)

api = FastAPI()
add_routes(api, fraud_app, path="/fraud")

# 启动：uvicorn worker_fraud:api --port 8002

# 调用时必须指定 thread_id
config = {"configurable": {"thread_id": "claim-12345"}}

# Worker 1
result1 = triage_app.invoke(input_data, config=config)

# Worker 2（使用相同 thread_id）
result2 = fraud_app.invoke(None, config=config)  # 自动恢复状态

from fastapi import FastAPI
from langserve import add_routes

app = FastAPI()

# 魔法：将任何 Runnable 变成 REST API
add_routes(
    app,
    your_agent_or_chain,
    path="/agent",
    enabled_endpoints=["invoke", "batch", "stream"]
)

# server.py
from fastapi import FastAPI
from langserve import add_routes
from insurance_swarm import app as swarm_app  # 之前定义的 LangGraph

api = FastAPI(
    title="Insurance Claims API",
    version="1.0",
    description="Distributed multi-agent insurance processing"
)

add_routes(
    api,
    swarm_app,
    path="/claims",
    enabled_endpoints=["invoke", "stream"]
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(api, host="0.0.0.0", port=8000)

# client.py
from langserve import RemoteRunnable

# 连接远程服务
remote_claims = RemoteRunnable("http://localhost:8000/claims")

# 像调用本地函数一样
result = remote_claims.invoke({
    "messages": [{"role": "user", "content": "报案编号 C9527"}],
    "claim_id": "",
    "risk_score": 0.0
})

print(result["messages"][-1]["content"])

for chunk in remote_claims.stream(input_data):
    print(chunk, end="", flush=True)

# 服务端定义 Pydantic 模型
class ClaimInput(BaseModel):
    claim_id: str
    description: str

# 客户端自动校验
remote_claims.invoke(ClaimInput(claim_id="C123", description="车损"))

# 只需设置环境变量，LangServe 自动上报追踪数据
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"]
)

@mcp.resource("user://profile/{user_id}")
def get_user_profile(user_id: str):
    return {"name": "Alice", "age": 30}

@mcp.tool()
def send_email(to: str, subject: str, body: str):
    """发送邮件"""
    # 实际发送逻辑
    return "Email sent"

@mcp.prompt()
def code_review_prompt(language: str):
    return f"Review this {language} code for security issues..."

# db_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3

# 初始化 MCP Server
mcp = FastMCP("DatabaseService")

@mcp.tool()
def query_db(sql: str) -> str:
    """执行只读 SQL 查询

    Args:
        sql: SELECT 语句（不允许 UPDATE/DELETE）

    Returns:
        查询结果的 JSON 字符串
    """
    # 安全检查
    if not sql.strip().lower().startswith("select"):
        return "Error: Only SELECT queries allowed"

    try:
        conn = sqlite3.connect("claims.db")
        cursor = conn.cursor()
        cursor.execute(sql)
        results = cursor.fetchall()
        conn.close()
        return str(results)
    except Exception as e:
        return f"Error: {str(e)}"

@mcp.resource("db://schema")
def get_schema():
    """获取数据库表结构"""
    conn = sqlite3.connect("claims.db")
    cursor = conn.cursor()
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchall()
    conn.close()
    return {"tables": tables}

if __name__ == "__main__":
    mcp.run()  # 默认通过 stdio 运行

# agent.py
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_react_agent
from langchain_openai import ChatOpenAI

async def main():
    # 1. 连接 MCP Server
    client = MultiServerMCPClient({
        "database": {
            "transport": "stdio",
            "command": "python",
            "args": ["db_server.py"]
        }
    })

    # 2. 自动发现并加载工具
    tools = await client.get_tools()
    print(f"Loaded tools: {[t.name for t in tools]}")
    # 输出: ['query_db']

    # 3. 创建 Agent
    llm = ChatOpenAI(model="gpt-4o")
    agent = create_react_agent(llm, tools)

    # 4. 运行
    result = agent.invoke({
        "input": "查询 claims 表中风险评分大于 80 的案件数量"
    })
    print(result["output"])

import asyncio
asyncio.run(main())

Agent: "需要查数据库"
  → 调用 query_db 工具
  → MCP Client 通过 stdio 发送 JSON-RPC 请求到 db_server.py
  → db_server.py 执行 SQL 并返回结果
  → Agent 收到结果并生成最终回答

# 安装
npm install -g @modelcontextprotocol/server-github

# 配置到 Agent
client = MultiServerMCPClient({
    "github": {
        "transport": "stdio",
        "command": "mcp-server-github",
        "env": {"GITHUB_TOKEN": "your-token"}
    }
})

场景：在线客服（分流 → 技术支持 / 售后）
方案：进程内 Swarm
理由：流程固定，状态简单，无需分布式

场景：保险理赔（每天 10 万单）
方案：分布式 Swarm (Redis Checkpointer) + LangServe
理由：需要水平扩展，异步处理

场景：企业 AI 中台（接入 50+ 业务系统）
方案：LangServe (Agent 服务) + MCP (工具标准化)
理由：跨团队协作，需要标准化接口

def router_with_ttl(state: InsuranceState):
    # 记录跳转次数
    hop_count = state.get("_hop_count", 0)

    if hop_count > 20:
        return END  # 强制终止

    # 正常路由逻辑
    next_node = normal_router(state)
    state["_hop_count"] = hop_count + 1
    return next_node

# 在关键节点保存快照，用于回滚
def processor_node(state: InsuranceState):
    # 处理前快照
    snapshot = state.copy()

    try:
        # 处理逻辑
        result = process_claim(state)
    except Exception as e:
        # 回滚
        return snapshot

    return result

from langsmith import traceable

@traceable(run_type="agent")
def triage_node(state: InsuranceState):
    # LangSmith 自动记录输入输出
    ...

def call_remote_agent_with_fallback(url: str, input_data: dict):
    try:
        # 尝试调用远程服务
        return RemoteRunnable(url).invoke(input_data)
    except Exception as e:
        # 降级到本地规则
        return fallback_rule_engine(input_data)

# 设置 TTL，避免内存泄漏
redis_client.setex(
    f"thread:{thread_id}",
    86400,  # 24 小时过期
    state_json
)

# 客户端批量调用，服务端自动并行处理
results = remote_agent.batch([
    {"input": "case 1"},
    {"input": "case 2"},
    {"input": "case 3"}
])

# 使用 connection pooling 复用进程
client = MultiServerMCPClient({
    "db": {
        "transport": "stdio",
        "command": "python",
        "args": ["db_server.py"],
        "pool_size": 5  # 预启动 5 个进程
    }
})

# 为不同 Agent 分配不同的 MCP Server 权限
triage_tools = await client.get_tools(server="readonly_db")
fraud_tools = await client.get_tools(server="full_access_db")

from pydantic import BaseModel, validator

class ClaimInput(BaseModel):
    claim_id: str

    @validator("claim_id")
    def validate_claim_id(cls, v):
        if not v.startswith("C"):
            raise ValueError("Invalid claim ID format")
        return v

from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter

# LangServe 服务添加限流
@app.post("/claims/invoke", dependencies=[Depends(RateLimiter(times=10, seconds=60))])