一、为什么需要 Agent？RAG 的边界在哪里？
1.1 RAG 的三大局限
| 局限 | 说明 | 用户痛点 |
|---|---|---|
| 静态知识 | 只能回答已存入向量库的内容 | “最新订单数据在哪？” → 无法回答 |
| 无操作能力 | 不能执行任何动作 | “帮我提交请假申请” → 无能为力 |
| 无状态推理 | 每次问答独立 | 多轮对话中丢失上下文 |
1.2 Agent 的核心能力
Agent = LLM + Memory + Tools + Planning
✅本质：让大模型像人一样思考 + 像程序一样执行
二、Qwen-Max 的 Function Calling 能力详解
阿里通义千问qwen-max / qwen-plus已原生支持Function Calling（函数调用），无需额外微调！
2.1 Function Calling 工作流程
- 用户提问 → LLM 判断是否需要调用工具；
- LLM 输出结构化 JSON，指定
tool_name和parameters； - 系统执行工具，获取结果；
- 将结果注入上下文，LLM 生成最终回答。
2.2 定义工具 Schema（OpenAI 兼容格式）
# tools_schema.py tools = [ { "type": "function", "function": { "name": "query_employee_performance", "description": "查询员工月度绩效等级和评分", "parameters": { "type": "object", "properties": { "employee_name": {"type": "string", "description": "员工姓名"}, "month": {"type": "string", "description": "查询月份，格式 YYYY-MM"} }, "required": ["employee_name", "month"] } } }, { "type": "function", "function": { "name": "submit_leave_request", "description": "提交请假申请", "parameters": { "type": "object", "properties": { "start_date": {"type": "string", "description": "开始日期 YYYY-MM-DD"}, "end_date": {"type": "string", "description": "结束日期 YYYY-MM-DD"}, "reason": {"type": "string", "description": "请假事由"} }, "required": ["start_date", "end_date", "reason"] } } } ]⚠️注意：Qwen 的 Function Calling 格式与 OpenAI 高度兼容，可直接复用生态工具。
三、构建自定义 Tools（连接企业系统）
3.1 数据库查询工具（SQLAlchemy）
# tools/db_tool.py from sqlalchemy import create_engine, text import os class DBQueryTool: def __init__(self): self.engine = create_engine(os.getenv("DB_URL")) def query_employee_performance(self, employee_name: str, month: str) -> str: """安全查询：防止 SQL 注入""" with self.engine.connect() as conn: result = conn.execute( text(""" SELECT performance_level, score FROM employee_performance WHERE name = :name AND month = :month """), {"name": employee_name, "month": month} ).fetchone() if result: return f"{employee_name} 在 {month} 的绩效等级为 {result[0]}，评分为 {result[1]}" else: return "未找到相关绩效记录"✅安全实践：
- 使用参数化查询；
- 限制可查询字段；
- 添加权限校验（后文详述）。
3.2 REST API 调用工具（审批/报销）
# tools/api_tool.py import httpx class ApprovalAPITool: def __init__(self, base_url: str, token: str): self.client = httpx.Client(base_url=base_url, headers={"Authorization": f"Bearer {token}"}) def submit_leave_request(self, start_date: str, end_date: str, reason: str) -> str: response = self.client.post("/api/v1/leave", json={ "start_date": start_date, "end_date": end_date, "reason": reason }) if response.status_code == 201: return f"请假申请已提交，单号：{response.json()['id']}" else: return f"提交失败：{response.text}"3.3 代码执行工具（数据计算/报告生成）
⚠️高风险！需严格沙箱隔离
# tools/code_tool.py import subprocess import tempfile import os def safe_execute_python(code: str, timeout: int = 10) -> str: """在临时文件中执行 Python 代码（仅限可信用户）""" if "os" in code or "subprocess" in code or "__import__" in code: return "禁止使用危险模块" with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: f.write(code) temp_path = f.name try: result = subprocess.run( ["python", temp_path], capture_output=True, text=True, timeout=timeout ) return result.stdout if result.returncode == 0 else f"Error: {result.stderr}" finally: os.unlink(temp_path)🔒生产建议：
- 仅对管理员开放；
- 使用 Docker 容器隔离；
- 限制 CPU/内存。
四、LangChain Agent 核心实现
4.1 注册 Tools 到 Agent
# agent_builder.py from langchain.agents import Tool, initialize_agent from langchain_community.chat_models import ChatOpenAI # Qwen 兼容 OpenAI 接口 def create_qwen_agent(tools: list): # 初始化 Qwen（通过 DashScope 兼容 OpenAI） llm = ChatOpenAI( model="qwen-max", openai_api_key=os.getenv("QWEN_API_KEY"), openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1" ) # 包装自定义工具 langchain_tools = [ Tool(name="query_employee_performance", func=db_tool.query_employee_performance, description="查询员工绩效"), Tool(name="submit_leave_request", func=api_tool.submit_leave_request, description="提交请假"), Tool(name="execute_python", func=safe_execute_python, description="执行安全的Python代码") ] # 创建 ReAct Agent（支持推理+行动） agent = initialize_agent( langchain_tools, llm, agent="react-chat", # 支持多轮思考 verbose=True, handle_parsing_errors=True ) return agent✅ReAct 框架：
- Thought：思考下一步做什么；
- Action：调用哪个工具；
- Observation：工具返回结果；
- Answer：最终回答。
4.2 融合 RAG 作为背景知识
# rag_enhanced_agent.py def create_rag_enhanced_agent(rag_retriever, tools): llm = get_qwen_llm() # 自定义 Agent Prompt，注入 RAG 上下文 prompt = """ 你是一个企业智能助手，可以调用工具或参考知识库回答问题。 知识库内容： {rag_context} 当前可使用的工具： {tools} 请按以下步骤思考： 1. 是否需要从知识库获取信息？ 2. 是否需要调用工具执行操作？ 3. 综合信息生成最终回答。 问题：{input} """ def get_rag_context(question: str) -> str: docs = rag_retriever.invoke(question) return "\n".join([d.page_content for d in docs[:2]]) # 构建链 chain = ( {"input": RunnablePassthrough()} | {"rag_context": lambda x: get_rag_context(x["input"]), "input": RunnablePassthrough()} | PromptTemplate.from_template(prompt) | llm | StrOutputParser() ) return chain💡关键设计：
- 先检索 RAG 获取制度/流程；
- 再决定是否调用工具执行具体操作。
五、多轮对话与短期记忆
5.1 使用 ConversationBufferMemory
from langchain.memory import ConversationBufferMemory memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) agent = initialize_agent( tools, llm, agent="conversational-react-description", memory=memory, verbose=True ) # 示例对话 agent.run("我是张三") agent.run("我上月绩效如何？") # Agent 自动关联“张三”5.2 会话隔离（多租户）
# 每个用户/会话独立 memory user_memories = {} def get_user_memory(user_id: str): if user_id not in user_memories: user_memories[user_id] = ConversationBufferMemory( memory_key="chat_history", return_messages=True, output_key="output" ) return user_memories[user_id]✅避免信息泄露：不同用户对话完全隔离。
六、企业级安全控制
6.1 工具级权限控制
# permission_checker.py USER_ROLES = { "zhangsan": ["read_performance", "submit_leave"], "manager": ["read_performance", "approve_leave", "run_report"] } def check_permission(user_id: str, tool_name: str) -> bool: role = get_user_role(user_id) required_perm = TOOL_PERMISSIONS.get(tool_name) return required_perm in USER_ROLES.get(role, [])6.2 敏感操作二次确认
# 在 Tool 执行前插入确认 if tool_name == "submit_leave_request": return "即将提交请假申请，确认吗？(是/否)"🔐安全原则：
- 最小权限原则；
- 所有写操作需确认；
- 审计日志全记录。
七、完整工作流演示
场景：生成销售分析报告
用户：
“请生成 2025 年 Q3 华东区销售分析报告”
Agent 执行流程：
- Thought：需要销售数据 + 报告模板
- Action：调用
query_sales_data(region="华东", quarter="Q3") - Observation：返回 CSV 数据
- Action：调用
retrieve_rag_document("销售报告模板") - Observation：返回 Markdown 模板
- Action：调用
execute_python(code="用 matplotlib 画柱状图...") - Observation：生成 chart.png
- Answer：整合数据、图表、模板，输出完整报告
✅全程无需人工干预，真正“能办事”
八、部署架构与性能优化
8.1 生产部署架构
8.2 性能优化策略
| 问题 | 解决方案 |
|---|---|
| LLM 调用慢 | 缓存高频工具调用结果 |
| 多步执行超时 | 异步任务队列（Celery） |
| 内存泄漏 | 限制每会话最大轮数（如 10 轮） |
| 成本高 | 对简单问题降级到纯 RAG |
九、评估与监控
9.1 关键指标
| 指标 | 目标 |
|---|---|
| 工具调用成功率 | >95% |
| 多步任务完成率 | >80% |
| 平均响应时间 | <3s（简单） / <10s（复杂） |
| 安全拦截率 | 100%（越权操作） |
9.2 审计日志示例
{ "timestamp": "2025-12-22T14:30:00", "user_id": "zhangsan", "query": "提交明天请假", "tools_called": ["submit_leave_request"], "parameters": {"start_date": "2025-12-23", ...}, "result": "申请成功，单号 LEAVE-20251223-001", "status": "success" }十、总结：从问答到办事，AI 的终极形态
| 能力 | RAG | RAG + Agent |
|---|---|---|
| 回答制度问题 | ✅ | ✅ |
| 查询实时数据 | ❌ | ✅ |
| 执行业务操作 | ❌ | ✅ |
| 多步复杂任务 | ❌ | ✅ |
| 主动规划 | ❌ | ✅ |
未来已来：
- RAG 是眼睛（看文档），Agent 是手脚（做事情）；
- 企业 AI 的竞争，不再是“能不能答”，而是“能不能办”；
- 谁先构建可操作的 AI 助手，谁就掌握效率革命的钥匙。
