苏三公子1. 多Agent协作的核心价值与应用场景
在AI技术快速发展的今天，多Agent协作系统正成为解决复杂问题的利器。这种架构将一个庞大任务拆解为多个子任务，由不同专业能力的Agent分工合作完成。就像一支高效的项目团队，每个成员各司其职又紧密配合。
典型应用场景包括：
- 智能数据处理流水线：如图像识别（Yolo）→文本提取（qwen-vl）→信息整合（LLM）的自动化流程
- 内容生成工作流：从素材收集、内容创作到排版设计的全自动PPT生成
- 复杂决策系统：金融分析中的数据采集、指标计算和风险评估的链式处理
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 案例解析：PE平均值计算系统设计
2.1 需求拆解与技术选型
我们需要实现一个查询多家公司PE值并计算平均值的系统。传统单Agent方案存在以下痛点：
- 单一模型难以同时精通数据查询和精确计算
- 错误处理逻辑复杂，容错性差
- 任务流程难以复用和扩展
解决方案架构：
mermaid复制graph TD
A[用户请求] --> B[Supervisor]
B -->|查询需求| C[Researcher]
C --> D[数据返回]
D --> B
B -->|计算需求| E[Coder]
E --> F[结果返回]
F --> B
B -->|完成| G[Finish]
2.2 核心组件实现细节
工具层开发
python复制@tool
def web_search(query: str):
"""专业金融数据查询工具
Args:
query: 查询语句，如"阿里巴巴 PE"
Returns:
str: 公司名称(股票代码) PE: 值
Example:
>>> web_search("腾讯 PE")
'腾讯控股(0700) PE: 18.2'
"""
# 模拟数据源，实际应接入金融API
data_map = {
"阿里": "阿里巴巴(BABA) PE: 15.5",
"腾讯": "腾讯控股(0700) PE: 18.2",
"百度": "百度(BIDU) PE: 11.8"
}
return data_map.get(query.strip(), "未找到数据")
@tool
def python_calculator(code: str):
"""安全计算执行器
Args:
code: 合法Python表达式
Returns:
str: 计算结果或错误信息
Example:
>>> python_calculator("(15.5+18.2+11.8)/3")
'计算结果: 15.166666666666666'
"""
try:
# 安全验证可加入AST解析检查
return f"计算结果: {eval(code)}"
except Exception as e:
return f"计算错误: {str(e)}"
Agent节点实现
python复制def create_agent(llm, tools, system_prompt):
"""Agent工厂函数
Args:
llm: 基础大模型
tools: 可用工具列表
system_prompt: 角色定义
Returns:
function: 可执行agent函数
"""
def _agent(state: dict):
bound_llm = llm.bind_tools(tools)
messages = [SystemMessage(content=system_prompt)] + state["messages"]
response = bound_llm.invoke(messages)
results = [response]
# 工具调用处理
for tool_call in response.tool_calls:
tool = next((t for t in tools if t.name == tool_call["name"]), None)
if tool:
output = tool.invoke(tool_call["args"])
results.append(ToolMessage(
content=str(output),
name=tool_call["name"],
tool_call_id=tool_call["id"]
))
return {"messages": results}
return _agent
# 专业化Agent实例
researcher = create_agent(
llm=llm,
tools=[web_search],
system_prompt="""你是一名金融数据分析师。职责：
1. 精确查询要求的财务指标
2. 返回原始数据，不做任何计算
3. 若数据不存在返回"未找到""""
)
coder = create_agent(
llm=llm,
tools=[python_calculator],
system_prompt="""你是一名计算工程师。职责：
1. 根据提供的数据编写计算代码
2. 确保代码语法正确
3. 返回完整计算表达式和结果"""
)
2.3 工作流编排引擎
python复制class AgentState(TypedDict):
messages: List[BaseMessage]
next: str
def build_workflow():
"""构建多Agent协作图"""
workflow = StateGraph(AgentState)
# 节点注册
workflow.add_node("Researcher", researcher)
workflow.add_node("Coder", coder)
workflow.add_node("Supervisor", supervisor)
workflow.add_node("Finish", finish_agent)
# 边缘逻辑
workflow.add_edge("Researcher", "Supervisor")
workflow.add_edge("Coder", "Supervisor")
workflow.add_edge("Finish", END)
# 条件路由
workflow.add_conditional_edges(
"Supervisor",
lambda state: state["next"],
{
"Researcher": "Researcher",
"Coder": "Coder",
"FINISH": "Finish"
}
)
workflow.set_entry_point("Supervisor")
return workflow.compile()
3. 关键技术与优化策略
3.1 结构化输出控制
python复制class RouteDecision(BaseModel):
next_step: Literal["Researcher", "Coder", "FINISH"] = Field(
...,
description="""下一步执行节点选择规则：
- Researcher: 需要查询新数据时
- Coder: 需要进行数学计算时
- FINISH: 任务完成时"""
)
def supervisor(state):
"""智能路由决策"""
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一名资深项目经理，负责任务分配"),
MessagesPlaceholder(variable_name="messages"),
SystemMessage(content="根据当前进度，决定下一步执行节点")
])
chain = prompt | llm.with_structured_output(RouteDecision)
decision = chain.invoke(state)
return {"next": decision.next_step}
3.2 性能优化实践
-
Agent复用策略：
- 对高频调用的Agent保持长连接
- 实现Agent池化减少初始化开销
-
缓存机制：
python复制from functools import lru_cache
@lru_cache(maxsize=100)
def cached_web_search(query: str):
return web_search(query)
- 超时控制：
python复制from concurrent.futures import ThreadPoolExecutor, TimeoutError
def safe_agent_call(agent, state, timeout=30):
with ThreadPoolExecutor() as executor:
future = executor.submit(agent, state)
try:
return future.result(timeout=timeout)
except TimeoutError:
return {"error": "处理超时"}
4. 生产环境注意事项
4.1 错误处理规范
- 工具级验证：
python复制@tool
def safe_calculator(code: str):
"""增强版安全计算器"""
ALLOWED_IMPORTS = {'math', 'statistics'}
try:
# 语法检查
ast.parse(code)
# 危险操作检查
for node in ast.walk(ast.parse(code)):
if isinstance(node, ast.Import):
raise ValueError("禁止导入模块")
return eval(code, {'__builtins__': None}, ALLOWED_IMPORTS)
except Exception as e:
return f"安全错误: {str(e)}"
- Agent级监控：
python复制def monitored_agent(agent):
def wrapper(state):
start = time.time()
try:
result = agent(state)
log_metric("success", 1)
return result
except Exception as e:
log_metric("error", 1)
return {"error": str(e)}
finally:
log_metric("latency", time.time()-start)
return wrapper
4.2 安全防护措施
- 输入消毒：
python复制def sanitize_input(text: str):
"""防止提示词注入"""
BLACKLIST = ["system", "import", "exec"]
return "".join(c for c in text if c.isprintable() and
not any(word in text.lower() for word in BLACKLIST))
- 权限控制：
python复制class RBACAgent:
def __init__(self, role, permissions):
self.role = role
self.allowed_tools = permissions.get(role, [])
def __call__(self, state):
if state.get('tool') not in self.allowed_tools:
return {"error": "权限拒绝"}
# ...原有逻辑
5. 扩展应用场景
5.1 多模态处理流水线
mermaid复制graph LR
A[图像输入] --> B(OCR Agent)
B --> C(文本摘要 Agent)
C --> D(多语言翻译 Agent)
D --> E[结果输出]
5.2 分布式任务调度
python复制from celery import Celery
app = Celery('agents', broker='redis://localhost:6379/0')
@app.task
def async_agent_run(agent_name, state):
agent = get_agent(agent_name) # 从注册中心获取
return agent(state)
在实际项目中，我们通过这种架构实现了：
- 金融报表自动分析系统（数据采集→指标计算→报告生成）
- 智能客服工单处理（意图识别→信息查询→解决方案生成）
- 跨境电商商品管理（多语言翻译→合规检查→定价优化）