从聊天到执行：基于LangChain构建智能体WorkBuddy的架构与实现

1. 项目概述：一个“会干活”的AI伙伴

最近在折腾AI应用落地的朋友，估计都绕不开一个核心痛点：如何让大语言模型从一个“博学的聊天对象”，真正变成一个能帮你“干脏活累活”的可靠伙伴？这就是“WorkBuddy”这个项目想解决的根本问题。它不是一个简单的聊天机器人，而是一个旨在深度理解你的身份、习惯和任务上下文，并主动替你执行具体操作的智能体。

简单来说，传统的AI助手，你问它“今天天气怎么样？”，它会告诉你天气。但WorkBuddy的思路是，当你早上说“今天有什么安排？”时，它不仅能基于日历告诉你上午十点有会，还能根据你“总是提前15分钟到会议室”的习惯，在9:45提醒你动身，甚至自动帮你把会议资料从云盘下载到本地，并打开相关的演示文稿。它的目标是从被动应答的“我是谁”（身份识别与记忆），跃迁到主动服务的“帮我干活”（任务理解与自动化执行）。

这个项目适合所有对AI应用开发、智能体构建以及自动化工作流感兴趣的朋友，无论是想提升个人效率的极客，还是希望为企业内部打造智能助手的开发者。接下来，我会拆解实现这样一个“WorkBuddy”的核心设计思路、关键技术栈，并分享从零搭建原型过程中那些必须注意的“坑”。

2. 核心架构设计：让AI拥有“手”和“记忆”

要实现从认知到执行的跨越，系统的架构设计是关键。你不能只靠一个语言模型包打天下，它需要一套组合拳。

2.1 三层核心组件解析

一个完整的WorkBuddy系统，可以抽象为三层：感知与理解层、决策与规划层、执行与反馈层。

感知与理解层，这是“我是谁”的基础。它的核心是用户画像与上下文管理。你需要一个向量数据库来存储和检索用户的历史对话、行为偏好、个人资料。例如，用户提到过“我习惯用Markdown写周报”，这句话需要被解析、嵌入成向量，存起来。下次用户说“写周报”时，系统不仅能检索到“周报”这个任务，还能关联到“Markdown格式”这个偏好。这里常用的工具有ChromaDB、Pinecone或者Weaviate。关键在于设计好数据的嵌入和检索策略，比如将用户信息、对话历史、任务结果分开存储，通过元数据过滤提高检索精度。

决策与规划层，这是大脑。大语言模型在这里扮演核心角色，但它不是直接输出答案，而是输出一个可执行的计划。当用户说“帮我整理一下上周的项目资料”，模型需要拆解任务：1. 确定时间范围（上周一至周日）；2. 定位资料存储位置（可能是Google Drive的“ProjectX”文件夹）；3. 定义“整理”的具体操作（按日期重命名文件、生成摘要目录）；4. 识别所需工具（调用云盘API、调用文件处理API）。这个过程被称为“任务分解”或“思维链规划”。我们通常使用像GPT-4、Claude 3或者开源的Llama 3、DeepSeek等具有较强推理能力的模型，并通过精心设计的提示词来引导其进行结构化输出。

执行与反馈层，这是“手”。AI自己不能操作电脑，它需要通过工具调用来实现。你需要为模型提供一套“工具集”，比如：read_file、search_web、send_email、execute_shell_command（需极其谨慎）等。模型根据规划层的输出，决定调用哪个工具，并生成正确的调用参数。执行后，工具返回结果（成功或失败，附带数据），这个结果会反馈给模型，模型根据结果决定下一步是继续执行还是向用户请求澄清。这一层的实现框架，ReAct和LangChain是很好的起点，它们提供了将模型、工具、记忆连接起来的标准化模式。

注意：工具调用是安全风险最高的环节。绝对禁止开放高危工具（如无条件执行任意Shell命令、直接访问数据库写操作）给模型。必须实施严格的工具权限管控和参数校验。例如，文件删除工具只能删除特定临时目录下的文件；邮件发送工具需要二次确认或限制收件人域名。

2.2 状态管理与会话持久化

WorkBuddy需要记住跨会话的信息。这不仅仅是保存聊天记录那么简单，而是维护一个动态的会话状态。这个状态包括：当前正在执行的多步骤任务进度、已收集到的用户输入参数、临时生成的数据等。例如，用户说“订一张明天去上海的机票”，WorkBuddy进入订票流程，状态变为awaiting_departure_time。你回复“下午出发”，状态更新，并接着问“您偏好哪个航空公司？”。

实现上，可以将整个会话状态（一个复杂的Python字典或Pydantic模型）序列化后，与用户ID关联存储到数据库（如Redis用于快速会话缓存，PostgreSQL用于长期持久化）。每次交互开始前，先加载状态，这样模型就知道“对话进行到哪一步了”。LangChain的ConversationChain或自定义的StateGraph（在LangGraph中）是管理这类状态的强大工具。

3. 关键技术实现与工具链选型

理论讲完，我们落到具体的代码和工具上。我会以构建一个简单的、能处理“文件整理”和“信息查询”任务的WorkBuddy原型为例。

3.1 搭建基础环境与智能体核心

首先，我们选择LangChain作为智能体框架，因为它生态丰富，抽象得当。假设我们使用OpenAI的GPT-4作为核心模型。

# 环境准备 pip install langchain langchain-openai langchain-community chromadb python-dotenv

# core_agent.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain.prompts import PromptTemplate from langchain_community.tools import DuckDuckGoSearchRun # 假设我们自定义了一些工具，后面会定义 from custom_tools import FileSearchTool, SummarizeDocumentTool load_dotenv() # 1. 初始化大模型 llm = ChatOpenAI( model="gpt-4-turbo-preview", temperature=0.1, # 低温度保证任务执行的稳定性 api_key=os.getenv("OPENAI_API_KEY") ) # 2. 初始化记忆（用于存储对话历史） memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True, output_key="output" ) # 3. 定义工具集 search = DuckDuckGoSearchRun() file_search = FileSearchTool() summarizer = SummarizeDocumentTool() tools = [ Tool( name="Web Search", func=search.run, description="Useful for when you need to answer questions about current events or general knowledge. Input should be a clear search query." ), Tool( name="File Search", func=file_search.run, description="Useful for searching and retrieving information from the user's document repository. Input should be keywords related to the file content." ), Tool( name="Document Summarizer", func=summarizer.run, description="Useful for summarizing long documents. Input should be the path to a text file or the text content itself." ) ] # 4. 创建ReAct智能体 prompt = PromptTemplate.from_template(""" You are WorkBuddy, a helpful assistant that can perform tasks. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input Begin! Previous conversation history: {chat_history} Question: {input} Thought:{agent_scratchpad} """) agent = create_react_agent(llm, tools, prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, memory=memory, verbose=True, # 开发时打开，查看思考过程 handle_parsing_errors=True, max_iterations=5 # 防止死循环 ) # 运行示例 if __name__ == "__main__": response = agent_executor.invoke({"input": "帮我找一下上个月关于市场分析的文档，并总结核心观点。"}) print(response["output"])

这个代码搭建了一个最基本的、具备思考（Thought）和行动（Action）能力的智能体。它可以通过搜索工具找文件，然后用总结工具生成摘要。

3.2 实现自定义工具：以文件操作为例

上面用到的FileSearchTool和SummarizeDocumentTool需要我们自己实现。这里展示一个连接简单本地文件系统的搜索工具。

# custom_tools.py import os from pathlib import Path from typing import Type from langchain.tools import BaseTool from pydantic import BaseModel, Field from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings class FileSearchInput(BaseModel): query: str = Field(description="Keywords to search for within documents") directory: str = Field(default=".", description="Directory to search in") class FileSearchTool(BaseTool): name = "file_search" description = "Searches for content within text and PDF files in a specified directory." args_schema: Type[BaseModel] = FileSearchInput return_direct: bool = False # 结果返回给Agent继续处理 vector_store = None def _setup_vector_store(self, directory: str): """初始化或加载指定目录的文档向量库""" persist_path = f"./chroma_db/{os.path.basename(directory)}" embeddings = OpenAIEmbeddings() if os.path.exists(persist_path): # 加载已有库 self.vector_store = Chroma( persist_directory=persist_path, embedding_function=embeddings ) else: # 构建新库 documents = [] for ext in ['*.txt', '*.md', '*.pdf']: for file_path in Path(directory).rglob(ext): try: if file_path.suffix == '.pdf': loader = PyPDFLoader(str(file_path)) else: loader = TextLoader(str(file_path), encoding='utf-8') loaded_docs = loader.load() # 为每个文档添加来源元数据 for doc in loaded_docs: doc.metadata["source"] = str(file_path) doc.metadata["filename"] = file_path.name documents.extend(loaded_docs) except Exception as e: print(f"Error loading {file_path}: {e}") if documents: # 分割文本，便于检索 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) split_docs = text_splitter.split_documents(documents) # 创建向量存储 self.vector_store = Chroma.from_documents( documents=split_docs, embedding=embeddings, persist_directory=persist_path ) self.vector_store.persist() else: self.vector_store = None def _run(self, query: str, directory: str = ".") -> str: if not query: return "Please provide a search query." # 懒加载向量库 if self.vector_store is None: self._setup_vector_store(directory) if self.vector_store is None: return f"No indexable text or PDF files found in {directory}." # 执行相似性搜索 try: results = self.vector_store.similarity_search_with_relevance_scores(query, k=3) if not results: return "No relevant documents found for your query." response = "Here are the most relevant documents I found:\n" for i, (doc, score) in enumerate(results): response += f"\n{i+1}. **{doc.metadata.get('filename', 'Unknown')}** (Relevance: {score:.2f})\n" response += f" Source: {doc.metadata.get('source')}\n" response += f" Snippet: {doc.page_content[:200]}...\n" return response except Exception as e: return f"An error occurred during search: {str(e)}" async def _arun(self, query: str, directory: str = "."): raise NotImplementedError("Async operation not supported yet.")

这个工具做了几件关键事：1. 它接受查询和目录参数；2. 首次使用时，会加载目录下的文档并构建向量索引（这是一个耗时的初始化过程，实际产品中需要异步或后台处理）；3. 执行语义搜索，返回最相关的文档片段及其来源。这比简单文件名搜索强大得多，因为它理解内容。

3.3 用户画像与长期记忆的实现

让WorkBuddy记住“我是谁”，我们需要一个用户画像系统。一个简单的实现可以基于向量数据库存储用户相关的所有信息片段。

# user_profile.py from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document from datetime import datetime import json class UserProfileManager: def __init__(self, user_id: str): self.user_id = user_id self.profile_store_path = f"./user_profiles/{user_id}" self.embeddings = OpenAIEmbeddings() self._init_store() def _init_store(self): """初始化或加载用户的向量化记忆库""" try: self.vector_store = Chroma( persist_directory=self.profile_store_path, embedding_function=self.embeddings ) except: # 首次创建 self.vector_store = Chroma.from_documents( documents=[], embedding=self.embeddings, persist_directory=self.profile_store_path ) def add_memory(self, memory_text: str, memory_type: str = "conversation", metadata: dict = None): """添加一段记忆到用户画像""" if metadata is None: metadata = {} metadata.update({ "user_id": self.user_id, "type": memory_type, "timestamp": datetime.now().isoformat() }) doc = Document(page_content=memory_text, metadata=metadata) self.vector_store.add_documents([doc]) self.vector_store.persist() def query_memory(self, query: str, memory_type: str = None, k: int = 3): """查询与当前问题相关的用户记忆""" filter_dict = {"user_id": self.user_id} if memory_type: filter_dict["type"] = memory_type results = self.vector_store.similarity_search_with_relevance_scores( query, k=k, filter=filter_dict ) return results def get_user_context(self, current_query: str) -> str: """获取与当前查询相关的用户上下文，用于拼接到提示词中""" relevant_memories = self.query_memory(current_query) if not relevant_memories: return "No specific user context found." context_str = "Relevant user context from past interactions:\n" for i, (doc, score) in enumerate(relevant_memories): context_str += f"- {doc.page_content} (from {doc.metadata.get('type', 'unknown')})\n" return context_str # 使用示例 if __name__ == "__main__": user_manager = UserProfileManager("user_123") # 添加一些记忆 user_manager.add_memory( "The user prefers to receive summaries in bullet points.", memory_type="preference" ) user_manager.add_memory( "The user is working on a project called 'Phoenix' related to market analysis.", memory_type="project" ) # 查询 context = user_manager.get_user_context("How should I format the report?") print(context)

这样，每次用户交互时，我们可以先调用get_user_context获取相关历史信息，并将其作为系统提示词的一部分，让模型在了解用户背景的情况下进行回复和规划。

4. 提示词工程：引导AI正确规划与执行

智能体的“智商”很大程度上取决于提示词。对于WorkBuddy，我们需要设计多阶段的提示词。

4.1 系统提示词设计

系统提示词定义了AI的角色、能力和行为规范。这是最核心的部分。

你是一个名为WorkBuddy的智能助手。你的核心目标是理解用户需求，并主动、准确地完成任务。 **你的身份与原则：** 1. 你是用户的专属助手，熟知用户的偏好和历史（上下文会提供给你）。 2. 你拥有工具使用能力。在回答问题或执行任务时，应优先考虑是否可以通过调用工具更高效、准确地完成。 3. 对于复杂任务，你必须先制定分步计划，然后逐步执行。每次只执行一步，观察结果后再决定下一步。 4. 如果工具执行失败，或结果不明确，不要猜测。向用户澄清或尝试替代方案。 5. 始终以帮助用户完成具体工作为目标，回答应简洁、务实，避免冗长的理论阐述。 **可用的工具：** {tools} **当前用户上下文：** {user_context} **对话历史：** {chat_history} 请严格按照以下格式输出： Thought: 分析用户请求，结合上下文和历史。决定是否需要使用工具，以及使用哪个工具。 Action: 需要使用的工具名，如果不使用工具，则为 `None`。 Action Input: 工具的输入参数，如果不使用工具，则为 `None`。 Observation: 工具返回的结果（第一步先留空）。 ...（重复 Thought/Action/Action Input/Observation 直到任务完成或无需更多工具） Thought: 我现在有足够信息给出最终答案。 Final Answer: 对用户的最终回复，汇总结果或确认任务完成。

这个提示词明确了ReAct格式，并强调了规划、工具优先和基于上下文的行动。

4.2 任务分解与参数提取提示词

对于模糊的用户指令，我们需要一个专门的步骤来澄清和分解。这可以通过一个独立的“规划器”LLM调用来实现。

# planner_prompt.py from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field from typing import List class Subtask(BaseModel): description: str = Field(description="对该子任务的清晰描述") tool: str = Field(description="完成此任务所需的工具名称，如 'FileSearchTool', 'Web Search', 或 'None'") parameters: dict = Field(default_factory=dict, description="调用工具所需的参数") class TaskPlan(BaseModel): main_goal: str = Field(description="用户的主要目标") subtasks: List[Subtask] = Field(description="为达成目标所需的子任务序列") needs_clarification: bool = Field(description="是否需要向用户询问更多信息") clarification_question: str = Field(default="", description="如果需要澄清，要问什么问题") planner_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个任务规划专家。请将用户的模糊请求分解为具体、可执行的子任务步骤。 考虑可用的工具：{tool_descriptions}。 如果用户请求缺少必要信息（如时间、地点、具体文件名），请标记需要澄清，并生成一个清晰的问题。 输出必须是有效的JSON格式。"""), ("human", "用户请求：{user_input}\n当前已知用户上下文：{user_context}") ]) def create_plan(user_input: str, user_context: str, llm): parser = JsonOutputParser(pydantic_object=TaskPlan) full_prompt = planner_prompt.format_parts( user_input=user_input, user_context=user_context, tool_descriptions=", ".join([f"{t.name}: {t.description}" for t in tools]) # 假设tools已定义 ) message = full_prompt.to_messages() response = llm.invoke(message) plan_dict = parser.parse(response.content) return TaskPlan(**plan_dict)

这个规划器会先于主智能体运行。如果needs_clarification为True，WorkBuddy会先问用户那个问题。否则，它就获得了一个清晰的子任务列表，可以按顺序执行或交由智能体动态调整。

5. 安全、隐私与工程化考量

将AI智能体投入实际使用，尤其是处理用户文件和数据时，安全和隐私是生命线。

5.1 工具执行沙箱与权限控制

绝对不能允许AI直接、无限制地调用系统命令或访问所有文件。必须实现一个安全层。

# security_layer.py import subprocess import os from pathlib import Path class SafeCommandTool: """一个安全的、受限的Shell命令执行工具""" allowed_commands = { 'list_dir': {'cmd': ['ls', '-la'], 'cwd': None}, 'current_date': {'cmd': ['date'], 'cwd': None}, # 可以定义更多白名单命令 } allowed_directories = ['/tmp/workbuddy', os.path.expanduser('~/workbuddy_safe')] @staticmethod def run(command_name: str, args: list = None, working_dir: str = None): if command_name not in SafeCommandTool.allowed_commands: return f"Error: Command '{command_name}' is not in the allowed list." config = SafeCommandTool.allowed_commands[command_name].copy() cmd_list = config['cmd'] if args: cmd_list.extend(args) # 限制工作目录 safe_cwd = working_dir if working_dir in SafeCommandTool.allowed_directories else config.get('cwd') if safe_cwd: Path(safe_cwd).mkdir(parents=True, exist_ok=True) try: result = subprocess.run( cmd_list, cwd=safe_cwd, capture_output=True, text=True, timeout=10 ) if result.returncode == 0: return result.stdout else: return f"Command failed with error: {result.stderr}" except subprocess.TimeoutExpired: return "Error: Command timed out." except Exception as e: return f"Error executing command: {str(e)}"

这个工具只允许执行预定义的白名单命令，并在指定目录下运行。对于文件操作，同样需要路径白名单或沙箱机制。

5.2 数据隐私与匿名化处理

如果WorkBuddy需要处理包含个人身份信息的数据，必须在发送给外部AI API（如OpenAI）前进行匿名化。

# privacy_filter.py import re class PrivacyFilter: patterns = { 'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'phone_cn': r'\b1[3-9]\d{9}\b', # 简单中国手机号匹配 'id_card': r'\b[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[0-9Xx]\b' # 简单身份证号匹配 } @staticmethod def anonymize_text(text: str): anonymized = text for ptype, pattern in PrivacyFilter.patterns.items(): anonymized = re.sub(pattern, f'[{ptype}_redacted]', anonymized) return anonymized @staticmethod def should_send_to_external_api(text: str, user_consent: bool): """决定文本是否可发送外部API""" if not user_consent: return False, "No user consent for external API." # 检查是否包含高敏感信息（可自定义规则） high_sensitivity_pattern = r'(password|token|key|secret)\s*[:=]\s*\S+' if re.search(high_sensitivity_pattern, text, re.IGNORECASE): return False, "Text contains high-sensitivity information." # 匿名化后发送 safe_text = PrivacyFilter.anonymize_text(text) return True, safe_text

在调用LLM API前，先通过should_send_to_external_api检查并处理文本。同时，所有本地存储的用户数据（如向量库）应进行加密。

5.3 错误处理与鲁棒性增强

智能体在复杂环境中必然会出错。健壮的错误处理机制必不可少。

# error_handling.py from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import openai class RobustAgentExecutor: def __init__(self, agent_executor): self.agent = agent_executor @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry=retry_if_exception_type((openai.APITimeoutError, openai.APIConnectionError)) ) def invoke_with_retry(self, input_dict): """带重试的调用，主要处理网络或API瞬时错误""" try: return self.agent.invoke(input_dict) except Exception as e: # 解析智能体特定的错误，如工具调用失败、输出解析错误 error_msg = str(e) if "Could not parse LLM output" in error_msg: # 模型输出不符合格式，可能是复杂问题导致混乱 return { "output": "I encountered an issue while planning the steps. Could you please rephrase your request or break it down into simpler tasks?" } elif "Tool not found" in error_msg: return { "output": "I tried to use a tool that isn't available. Let me try a different approach." } else: # 其他未预见的错误，记录并返回友好信息 print(f"Unhandled agent error: {error_msg}") return { "output": "An unexpected error occurred. Please try again or contact support if the problem persists." }

此外，要为每个工具调用设置超时，并为智能体的“思考循环”设置最大迭代次数，防止陷入死循环消耗资源。

6. 部署与迭代：从原型到可用产品

让WorkBuddy真正可用，需要把它包装成一个服务。

6.1 后端API服务搭建

使用FastAPI可以快速构建一个提供Web服务的后端。

# main.py (FastAPI 后端) from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import Optional import uuid from agent_core import get_agent_for_user # 假设这是封装好的智能体获取函数 from user_profile import UserProfileManager app = FastAPI(title="WorkBuddy API") # 简单的内存会话存储（生产环境用Redis） user_sessions = {} class UserRequest(BaseModel): message: str session_id: Optional[str] = None class AgentResponse(BaseModel): reply: str session_id: str needs_follow_up: bool = False @app.post("/chat", response_model=AgentResponse) async def chat_with_buddy(request: UserRequest): # 获取或创建会话 session_id = request.session_id or str(uuid.uuid4()) if session_id not in user_sessions: user_sessions[session_id] = { "agent": get_agent_for_user(session_id), "profile": UserProfileManager(session_id) } session = user_sessions[session_id] agent = session["agent"] profile_manager = session["profile"] # 1. 更新用户画像（将本次对话作为记忆存储） profile_manager.add_memory( f"User said: {request.message}", memory_type="conversation" ) # 2. 获取相关用户上下文 user_context = profile_manager.get_user_context(request.message) # 3. 调用智能体 try: response = agent.invoke({ "input": request.message, "user_context": user_context }) final_output = response.get("output", "I didn't get a response.") # 4. 将助手的回复也作为记忆存储（可选） profile_manager.add_memory( f"WorkBuddy replied: {final_output}", memory_type="conversation" ) # 判断是否需要后续跟进（例如，任务未完成，在等待用户输入参数） needs_follow_up = "please provide" in final_output.lower() or "?" in final_output return AgentResponse( reply=final_output, session_id=session_id, needs_follow_up=needs_follow_up ) except Exception as e: raise HTTPException(status_code=500, detail=f"Agent processing failed: {str(e)}") @app.get("/session/{session_id}/memory") async def get_session_memory(session_id: str, query: Optional[str] = None): """获取特定会话的记忆（调试用）""" if session_id not in user_sessions: raise HTTPException(status_code=404, detail="Session not found") profile_manager = user_sessions[session_id]["profile"] if query: memories = profile_manager.query_memory(query) return {"memories": [{"content": doc.page_content, "meta": doc.metadata} for doc, _ in memories]} else: # 返回最近的一些记忆（简化处理） return {"message": "Provide a 'query' parameter to search memories."}

这个API提供了聊天接口和记忆查询接口。前端可以通过轮询或WebSocket与它交互。

6.2 前端简单界面示例

一个简单的Streamlit前端可以快速演示功能。

# streamlit_app.py import streamlit as st import requests import json API_BASE_URL = "http://localhost:8000" # 假设后端运行在此 st.title("🤖 WorkBuddy - Your AI Assistant") st.caption("From 'Who I am' to 'Get things done'") # 初始化会话状态 if "session_id" not in st.session_state: st.session_state.session_id = None if "messages" not in st.session_state: st.session_state.messages = [] # 显示历史消息 for msg in st.session_state.messages: with st.chat_message(msg["role"]): st.markdown(msg["content"]) # 聊天输入 if prompt := st.chat_input("What can WorkBuddy do for you?"): # 添加用户消息 st.session_state.messages.append({"role": "user", "content": prompt}) with st.chat_message("user"): st.markdown(prompt) # 调用后端API with st.chat_message("assistant"): with st.spinner("WorkBuddy is thinking..."): payload = { "message": prompt, "session_id": st.session_state.session_id } try: response = requests.post( f"{API_BASE_URL}/chat", json=payload, timeout=30 ) if response.status_code == 200: data = response.json() reply = data["reply"] st.session_state.session_id = data["session_id"] st.markdown(reply) st.session_state.messages.append({"role": "assistant", "content": reply}) # 如果AI需要更多信息，给出提示 if data.get("needs_follow_up"): st.info("WorkBuddy is waiting for more information to complete the task.") else: st.error(f"API Error: {response.text}") except requests.exceptions.RequestException as e: st.error(f"Failed to connect to WorkBuddy backend: {e}") # 侧边栏：会话管理 with st.sidebar: st.header("Session") if st.session_state.session_id: st.code(f"Session ID: {st.session_state.session_id[:8]}...") if st.button("New Session"): st.session_state.session_id = None st.session_state.messages = [] st.rerun() else: st.info("No active session. Start chatting to create one.")

6.3 监控、评估与持续改进

上线后，必须建立监控和评估体系。

1. 日志记录：记录每一次用户交互、AI的思考过程、工具调用（参数和结果）、最终输出。这用于调试和后续分析。
2. 关键指标：
   • 任务完成率：用户明确的任务中，有多少被成功执行？
   • 工具调用准确率：AI选择的工具是否合适？参数是否正确？
   • 用户满意度：通过简单的“赞/踩”按钮或后续调研收集。
   • 平均对话轮次：完成一个任务需要多少轮对话？轮次过多可能意味着规划能力或工具效率有问题。
3. 反馈循环：将失败的案例（如工具调用错误、用户不满意）加入一个评估数据集。定期用这个数据集微调模型的提示词，或者作为few-shot示例加入系统提示中。
4. 工具扩展：根据用户最常请求但当前无法完成的任务，优先级开发新的工具。例如，如果很多用户问“把这份PPT转换成PDF”，那就需要集成一个convert_ppt_to_pdf的工具。

7. 常见问题与实战排坑指南

在实际开发和测试中，我遇到了不少典型问题，这里总结一下。

7.1 智能体陷入循环或执行无关操作

这是ReAct模式最常见的问题。AI可能在一个步骤上不断重复，或者调用一系列不直接解决问题的工具。

解决方案：

• 严格限制迭代次数：如之前代码中的max_iterations=5，强制跳出。
• 改进提示词：在系统提示中强调“如果当前步骤无法推进任务，请停止并总结当前状况，向用户求助”。
• 后处理检查：在最终输出前，让另一个轻量级模型（或规则）检查本次对话是否实质性推进了用户目标。如果没有，触发一个重置或澄清。
• 示例学习：在提示词中加入几个成功完成多步任务的示例，让模型学会正确的规划模式。

7.2 工具描述不准确导致误用

如果工具的描述（description）过于宽泛或模糊，AI会错误地调用它。

踩坑案例：早期我给文件搜索工具的描述是“搜索文件”，结果用户问“明天的天气”，AI也去调用文件搜索。优化方案：将描述写得极其具体，限定使用场景。例如：“在用户指定的本地目录中，基于文档内容语义搜索相关的文本和PDF文件。适用于当用户想查找包含某些特定主题或关键词的文档时。不适用于查询实时信息、天气、新闻或网络搜索。”

7.3 处理模糊或开放式用户请求

用户常说“帮我做一下那个事”或“整理一下资料”。AI需要主动澄清。

实战技巧：除了前面提到的独立规划器，可以在主循环中设置一个“澄清状态”。当模型认为输入模糊时，它输出一个特定的动作，如Action: "ask_for_clarification"，然后前端捕获到这个特殊动作，弹出表单让用户填写缺失的字段（如时间、文件类型、具体操作）。这比让模型生成自然语言问题更结构化，也更容易处理。

7.4 上下文长度限制与记忆管理

长对话会耗尽模型的上下文窗口，导致忘记早期的关键信息。

解决方案组合拳：

1. 摘要记忆：定期（例如每10轮对话）让模型对之前的对话历史进行总结，将详细的对话压缩成几个要点，存入长期记忆（向量库），然后从当前对话上下文中移除旧的历史。这称为“对话摘要”。
2. 重要性评分：在存储记忆时，让模型对记忆的重要性打分（例如1-5分）。在检索时，优先召回高分记忆。用户明确说“记住这个”的信息，可以手动打高分。
3. 分层记忆：区分短期工作记忆（最近几轮对话）和长期档案记忆（向量库）。每次查询时，同时从两者中获取信息。

7.5 成本控制

频繁调用大模型和嵌入模型，成本可能快速上升。

优化策略：

• 缓存：对常见的、结果不变的查询（如“公司的核心价值观是什么”）的嵌入结果和AI回复进行缓存。
• 小模型分工：用便宜的小模型（如GPT-3.5 Turbo）处理简单的分类、路由或摘要任务，只在复杂的规划和创意生成时用大模型（如GPT-4）。
• 本地模型：对于嵌入任务，可以使用开源的本地模型（如all-MiniLM-L6-v2），避免调用OpenAI的Embedding API。对于简单的工具选择（基于描述），也可以训练一个小的分类模型。
• 预算监控：在代码中集成成本计算，对每个API调用记录token消耗，并设置每日/每月预算警报。

构建WorkBuddy这样的智能体是一个持续迭代的过程。从最简单的“问答机器人”到能“干活”的伙伴，中间需要不断地调试工具、优化提示词、完善安全策略。最深的体会是，不要指望一个超级提示词解决所有问题。可靠的智能体背后，是大量细致的工程工作：清晰的工具定义、稳健的错误处理、合理的状态管理以及持续的用户反馈学习。当你看到它第一次正确理解一个复杂任务，并自动调用一系列工具完成时，那种成就感是巨大的。