构建自己的 AI Agent · 第八篇：框架选型 & 生产部署 — 从手写代码到工业级服务

# 我们的代码 (第 1-4 篇)
class ConversationalAgent:
    def __init__(self):
        self.registry = ToolRegistry()       # → LangChain Tool
        self.conv = ConversationManager()     # → LangChain Memory
        self.memory = AgentMemory("./mem")    # → LangChain VectorStore
        self.max_iter = 10

    def run(self, query: str) -> str:
        for _ in range(self.max_iter):        # → LangGraph: AgentState loop
            response = self.client.chat.completions.create(...)
            if msg.tool_calls:                 # → LangGraph: conditional edge
                result = self.registry.execute(...)
            else:
                return msg.content


# LangChain/LangGraph 的等价写法
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)       # → 我们的 while 循环体
graph.add_node("tools", ToolNode(tools))  # → 我们的 registry.execute()
graph.add_edge("agent", "tools")          # → 我们 while 循环中的分支
graph.add_conditional_edges("agent", should_continue, {
    "continue": "tools",
    "end": END
})

# 我们的 AgentTeam (第 5 篇)
team = AgentTeam(members=[
    {"name": "研究员", "role": "收集信息"},
    {"name": "写作者", "role": "撰写报告"}
])
result = team.discuss("分析 AI 趋势", rounds=3)


# CrewAI 等价写法
from crewai import Agent, Task, Crew

researcher = Agent(role="研究员", goal="收集最新 AI 趋势信息")
writer = Agent(role="写作者", goal="撰写专业报告")

task1 = Task(description="搜索 AI 趋势", agent=researcher)
task2 = Task(description="撰写报告", agent=writer)

crew = Crew(agents=[researcher, writer], tasks=[task1, task2])
result = crew.kickoff()

# AutoGen 核心设计：ConversableAgent
from autogen import ConversableAgent, initiate_chats

assistant = ConversableAgent(
    name="assistant",
    system_message="你是助手",
    llm_config={"model": "gpt-4o"}
)

critic = ConversableAgent(
    name="critic",
    system_message="你是评论者，找出回答中的问题",
    llm_config={"model": "gpt-4o"}
)

# 两个 Agent 之间自动对话
initiate_chats([
    {"sender": assistant, "recipient": critic,
     "message": "写一个排序算法", "max_turns": 3}
])

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    return f"{city}: 晴, 22°C"

agent = Agent(
    name="助手",
    instructions="你是生活助手",
    tools=[get_weather]
)

result = Runner.run_sync(agent, "北京天气怎么样？")
print(result.final_output)

# agent_api.py — 生产级 Agent API 服务
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import asyncio
import uuid
import time
import logging
from contextlib import asynccontextmanager

# 会话存储 (生产环境用 Redis)
sessions: dict[str, dict] = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 启动时加载 Agent
    app.state.agent = SecureAgent(
        base_agent=MemoryAwareAgent(
            system_prompt="你是一个专业的 AI 助手。",
            registry=create_default_registry(),
            persist_dir="/data/agent_memory"
        ),
        sandbox=True
    )
    logger.info("Agent 服务已启动")
    yield
    # 关闭时清理
    logger.info("Agent 服务正在关闭")

app = FastAPI(title="AI Agent API", version="1.0.0", lifespan=lifespan)
app.add_middleware(CORSMiddleware, allow_origins=["*"],
                   allow_methods=["*"], allow_headers=["*"])

# ── 请求/响应模型 ──
class AgentRequest(BaseModel):
    query: str = Field(..., description="用户问题", max_length=10000)
    session_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    model: str = Field(default="auto", description="模型选择: auto/gpt-4o/gpt-4o-mini/o3-mini")
    stream: bool = Field(default=False)

class AgentResponse(BaseModel):
    session_id: str
    answer: str
    iterations: int
    tokens_used: int
    latency_ms: float
    model_used: str

# ── API 端点 ──
@app.post("/agent/chat", response_model=AgentResponse)
async def chat(request: AgentRequest):
    """Agent 对话端点"""
    start_time = time.time()
    logger.info(f"[{request.session_id}] 收到请求: {request.query[:100]}")

    # 获取或创建会话
    if request.session_id not in sessions:
        sessions[request.session_id] = {
            "created_at": time.time(),
            "messages": [],
            "token_total": 0
        }

    # 选择模型
    if request.model == "auto":
        tier = CostController().select_model(request.query)
        model_name = tier["model"]
    else:
        model_name = request.model

    # 执行 Agent
    agent = app.state.agent
    agent.agent.model = model_name

    try:
        answer = await asyncio.to_thread(agent.run, request.query)
    except Exception as e:
        logger.error(f"[{request.session_id}] Agent 执行失败: {e}")
        raise HTTPException(status_code=500, detail=str(e))

    latency = (time.time() - start_time) * 1000

    # 更新会话统计
    session = sessions[request.session_id]
    session["messages"].append({"role": "user", "content": request.query})
    session["messages"].append({"role": "assistant", "content": answer})

    logger.info(f"[{request.session_id}] 完成: {latency:.0f}ms")

    return AgentResponse(
        session_id=request.session_id,
        answer=answer,
        iterations=getattr(agent, '_last_iterations', 0),
        tokens_used=getattr(agent, '_last_tokens', 0),
        latency_ms=latency,
        model_used=model_name
    )

@app.get("/agent/sessions/{session_id}")
async def get_session(session_id: str):
    """获取会话信息"""
    if session_id not in sessions:
        raise HTTPException(status_code=404, detail="会话不存在")
    return sessions[session_id]

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {
        "status": "healthy",
        "active_sessions": len(sessions),
        "uptime": time.time() - start_time
    }

@app.get("/metrics")
async def metrics():
    """Prometheus 指标端点"""
    return {
        "agent_requests_total": total_requests,
        "agent_latency_p50": latency_histogram.get("p50", 0),
        "agent_latency_p95": latency_histogram.get("p95", 0),
        "agent_success_rate": success_rate,
        "agent_active_sessions": len(sessions),
    }


# ── 启动 ──
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, workers=4)

# 医学图像分析 Agent 配置
medical_agent = MemoryAwareAgent(
    system_prompt="""你是医学图像分析助手，专精于：

1. **分割模型推荐**：根据图像模态（CT/MRI/超声/X光）和目标器官推荐合适的模型
   - nnU-Net：通用分割基线（所有模态）
   - V-Net：3D 医学图像（CT/MRI 体积数据）
   - Attention U-Net：需要关注特定区域时
   - TransUNet：需要结合全局上下文时
   - SAM/MedSAM：零样本分割需求

2. **预处理建议**：
   - CT：窗宽窗位调整、HU 值归一化
   - MRI：偏置场校正（N4ITK）、多序列配准
   - 超声：散斑降噪、时间增益补偿

3. **训练建议**：
   - 损失函数选择（Dice/CrossEntropy/Focal/组合）
   - 数据增强策略（弹性形变对医学图像特别重要）
   - 小样本策略（迁移学习、少样本学习）""",

    registry=create_medical_registry(),
    persist_dir="./medical_agent_memory"
)

# 注册医学专用工具
@medical_agent.registry.register
def recommend_segmentation_model(
    modality: Literal["CT", "MRI", "Ultrasound", "X-ray"],
    organ: str,
    data_size: Literal["small", "medium", "large"] = "medium"
) -> dict:
    """根据图像模态和目标器官推荐分割模型"""
    recommendations = {
        ("CT", "liver"): ["nnU-Net", "V-Net"],
        ("MRI", "brain"): ["nnU-Net", "Attention U-Net", "TransUNet"],
        ("CT", "lung"): ["nnU-Net", "U-Net++"],
        ("Ultrasound", "heart"): ["Attention U-Net", "DeepLabV3+"],
    }
    key = (modality, organ.lower())

    if key in recommendations:
        models = recommendations[key]
        if data_size == "small":
            models = [m for m in models if m != "TransUNet"]
            models.append("nnU-Net (with transfer learning)")
        return {"recommended": models, "baseline": "nnU-Net"}

    return {"recommended": ["nnU-Net"], "rationale": "nnU-Net 作为通用基线适用于未见过的模态/器官组合"}


# 使用示例
result = medical_agent.run(
    "我有一组腹部 CT 数据，需要分割肝脏和肿瘤，数据量约 50 例。"
    "推荐合适的模型和预处理策略。"
)