开源 Agent 框架选型：看抽象边界，不只看示例

# ❌ 坏做法：依赖隐式 context 传递，任务链长了会丢数据
from crewai import Agent, Task, Crew
 
classifier = Agent(role='分类员', goal='对工单分类', backstory='工单分类专家')
retriever = Agent(role='检索员', goal='检索知识库', backstory='知识库检索专家')
writer = Agent(role='撰写员', goal='生成回复', backstory='客服回复撰写专家')
 
# 每个 Task 通过 context 引用前一个任务的输出
# 问题：当链超过两层，早期任务的输出可能被截断
classify_task = Task(description='分类工单: {ticket}', agent=classifier, expected_output='分类结果')
retrieve_task = Task(description='检索相关知识', agent=retriever, context=[classify_task], expected_output='知识条目')
write_task = Task(description='生成回复草稿', agent=writer, context=[retrieve_task], expected_output='回复草稿')
# write_task 拿不到 classify_task 的输出，工单 ID 丢了

# ✅ 好做法：用显式状态对象贯穿全流程
from crewai import Agent, Task, Crew
from pydantic import BaseModel
 
class TicketState(BaseModel):
    ticket_id: str
    user_info: dict
    category: str = ''
    knowledge: list[str] = []
    draft: str = ''
 
classifier = Agent(role='分类员', goal='对工单分类', backstory='工单分类专家')
retriever = Agent(role='检索员', goal='检索知识库', backstory='知识库检索专家')
writer = Agent(role='撰写员', goal='生成回复', backstory='客服回复撰写专家')
 
# 所有 Task 共享同一个 state 对象，字段显式传递
state = TicketState(ticket_id='TK-001', user_info={'name': '张三', 'level': 'VIP'})
classify_task = Task(description='分类工单', agent=classifier, expected_output='分类结果')
retrieve_task = Task(description='检索相关知识', agent=retriever, expected_output='知识条目')
write_task = Task(description='生成回复草稿', agent=writer, expected_output='回复草稿')
 
crew = Crew(agents=[classifier, retriever, writer], tasks=[classify_task, retrieve_task, write_task])
# 显式状态确保每个步骤都能读到完整上下文，不依赖隐式传递
result = crew.kickoff(inputs={'state': state.model_dump()})

# ❌ 坏做法：不设对话上限，完全依赖 LLM 自己判断何时收敛
from autogen import AssistantAgent, UserProxyAgent
 
drafter = AssistantAgent(name='起草员', system_message='你是一个工单回复起草员')
reviewer = AssistantAgent(name='审核员', system_message='审核回复质量，不满意就要求修改')
 
# 两个 Agent 自由对话，没有轮次和 token 限制
group_chat = autogen.GroupChat(
    agents=[drafter, reviewer],
    messages=[],
    # 没设 max_round，对话可能无限进行
)
manager = autogen.GroupChatManager(groupchat=group_chat)
# 遇到模糊工单时，reviewer 可能反复要求修改，token 失控

# ✅ 好做法：设置硬性对话上限 + token 预算回调
from autogen import AssistantAgent
import autogen
 
def token_budget_hook(messages, sender, config):
    """当对话轮次超限或 token 超标时强制终止"""
    if len(messages) >= 6:  # 最多 3 轮对话（每轮 2 条消息）
        return {'terminate': True, 'content': '对话达到上限，使用当前最佳版本'}
    total_tokens = sum(len(m.get('content', '')) for m in messages)
    if total_tokens > 4000:  # token 预算上限
        return {'terminate': True, 'content': 'Token 预算耗尽，使用当前最佳版本'}
    return {'terminate': False}
 
drafter = AssistantAgent(name='起草员', system_message='你是一个工单回复起草员')
reviewer = AssistantAgent(name='审核员', system_message='审核回复质量，最多提两轮修改意见')
 
group_chat = autogen.GroupChat(
    agents=[drafter, reviewer],
    messages=[],
    max_round=6,  # 硬性上限
)
# 注册 token 预算检查，超限自动降级
manager = autogen.GroupChatManager(
    groupchat=group_chat,
    # 对话终止条件不只看 LLM 输出，还要看资源消耗
)

# ❌ 坏做法：LangGraph 自建连接，和现有连接池互相抢资源
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
 
# AsyncPostgresSaver 内部会创建自己的连接池
# 和 SQLAlchemy 的连接池独立运行，两个池加起来超过 max_connections
checkpointer = AsyncPostgresSaver.from_conn_string(
    'postgresql://user:pass@localhost:5432/agent_db'
)
# 高并发时：两个连接池 + 应用连接 = 数据库连接耗尽

# ✅ 好做法：复用已有的连接池，统一管理连接资源
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from sqlalchemy.ext.asyncio import create_async_engine
 
# 复用应用已有的 async engine，共享连接池配置
engine = create_async_engine(
    'postgresql+asyncpg://user:pass@localhost:5432/agent_db',
    pool_size=10,
    max_overflow=20,
    # 整个应用只有一个连接池，LangGraph checkpointer 也从这里取连接
)
 
checkpointer = AsyncPostgresSaver(engine)
# 连接数可控，不会因为多了一个框架就多出一组连接

# ❌ 坏做法：工具函数内部持有框架特定的上下文对象
from langgraph.prebuilt import ToolNode
 
def search_knowledge(query: str, config: RunnableConfig) -> str:
    # 工具函数依赖 LangChain 的 RunnableConfig
    # 换框架后这个函数不可复用
    store = config['configurable']['vector_store']
    return store.similarity_search(query)
 
# ✅ 好做法：工具函数是纯函数，框架适配在最外层
def search_knowledge(query: str, store: VectorStore) -> str:
    # 纯函数，不依赖任何框架特定的上下文
    return store.similarity_search(query)
 
# 在 LangGraph 节点中适配
def knowledge_node(state: GraphState) -> dict:
    result = search_knowledge(state['query'], app_state.vector_store)
    return {'knowledge': result}

# ❌ 坏做法：状态定义嵌在框架的图定义里
from langgraph.graph import StateGraph
from typing import TypedDict
 
class GraphState(TypedDict):
    # 状态类型和图编排绑死，换框架要重写状态定义
    ticket_id: str
    category: str
    knowledge: list[str]
    draft: str
 
graph = StateGraph(GraphState)
 
# ✅ 好做法：状态是独立的领域模型，框架只是状态的消费者
from pydantic import BaseModel
 
class TicketContext(BaseModel):
    # 领域模型独立于任何框架
    ticket_id: str
    category: str = ''
    knowledge: list[str] = []
    draft: str = ''
 
# 框架适配层：把领域模型映射到框架状态
# 换框架时只需要重写这一层，领域模型不变
def to_graph_state(ctx: TicketContext) -> dict:
    return ctx.model_dump()

# ❌ 坏做法：只依赖框架的 verbose 模式做调试
# CrewAI
crew = Crew(..., verbose=True)
# verbose=True 在开发环境有用，生产环境日志不可结构化、不可检索
 
# ✅ 好做法：在工具函数层注入 OpenTelemetry span
from opentelemetry import trace
 
def search_knowledge(query: str, store: VectorStore) -> str:
    tracer = trace.get_tracer('ticket-agent')
    with tracer.start_as_current_span('knowledge_search') as span:
        span.set_attribute('query', query)
        results = store.similarity_search(query)
        span.set_attribute('result_count', len(results))
        return results
# 无论底层框架怎么换，可观测性始终跟着业务逻辑走

# ❌ 坏做法：依赖框架的默认重试策略
# AutoGen 默认不限制重试，LangGraph 默认不自动重试
# 框架的默认行为不一定适合你的业务场景
 
# ✅ 好做法：在工具函数层定义明确的重试和降级策略
from tenacity import retry, stop_after_attempt, fallback
 
@retry(stop=stop_after_attempt(3), reraise=True)
def call_external_api(endpoint: str, payload: dict) -> dict:
    # 业务层面的重试策略，和框架无关
    response = requests.post(endpoint, json=payload, timeout=10)
    response.raise_for_status()
    return response.json()
 
@fallback(fallback_value={'status': 'degraded', 'message': '使用缓存结果'})
def call_external_api_safe(endpoint: str, payload: dict) -> dict:
    return call_external_api(endpoint, payload)