【LLM实操系列07】Agent开发：构建自主AI智能体
在开始之前，建议先完成第04篇（理解ReAct概念）和第03篇（API调用）。你需要理解工具调用和思考-行动-观察循环的基本概念，并安装langchain及相关工具库。
10分钟实现第一个Agent
什么是Agent？
Agent = LLM + 工具 + 记忆 + 规划
最简Agent实现
# pip install langchain openai wikipedia duckduckgo-searchfromlangchain.agentsimportinitialize_agent,Toolfromlangchain.agentsimportAgentTypefromlangchain.chat_modelsimportChatOpenAIfromlangchain.toolsimportDuckDuckGoSearchRun# 1. 定义工具search=DuckDuckGoSearchRun()tools=[Tool(name="Search",func=search.run,description="搜索引擎，用于查找实时信息")]# 2. 创建Agentllm=ChatOpenAI(temperature=0)agent=initialize_agent(tools,llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,verbose=True)# 3. 执行任务result=agent.run("今天上海的天气如何？")print(result)ReAct框架：思考→行动→观察
基于第04篇的ReAct完整实现
# 注：第04篇介绍了ReAct的概念，这里实现完整的工具调用classReActAgent:"""ReAct模式Agent实现（完整版）"""def__init__(self,llm,tools):self.llm=llm self.tools={tool.name:toolfortoolintools}self.max_iterations=5defrun(self,task:str):"""执行任务"""prompt=f""" 任务：{task}可用工具：{self._format_tools()}使用以下格式： Thought: 我需要思考下一步做什么 Action: 工具名称 Action Input: 工具输入参数 Observation: 工具返回结果 ... (重复N次) Thought: 我知道最终答案了 Final Answer: 最终答案 开始： """foriinrange(self.max_iterations):# 获取LLM输出response=self.llm.predict(prompt)# 解析行动action=self._parse_action(response)ifaction["type"]=="Final Answer":returnaction["content"]# 执行工具tool=self.tools[action["tool"]]observation=tool.func(action["input"])# 添加观察结果prompt+=f""" Thought:{action["thought"]}Action:{action["tool"]}Action Input:{action["input"]}Observation:{observation}"""return"达到最大迭代次数"def_parse_action(self,text:str):"""解析LLM输出"""if"Final Answer:"intext:return{"type":"Final Answer","content":text.split("Final Answer:")[-1].strip()}# 提取Action和Inputlines=text.strip().split('\n')thought=""action=""action_input=""forlineinlines:ifline.startswith("Thought:"):thought=line[8:].strip()elifline.startswith("Action:"):action=line[7:].strip()elifline.startswith("Action Input:"):action_input=line[13:].strip()return{"type":"Action","thought":thought,"tool":action,"input":action_input}工具开发与集成
1. 自定义工具
fromtypingimportOptional,TypefrompydanticimportBaseModel,Fieldfromlangchain.toolsimportBaseToolimportrequestsclassWeatherInput(BaseModel):"""天气查询输入"""city:str=Field(description="城市名称")classWeatherTool(BaseTool):"""天气查询工具"""name="weather"description="查询指定城市的天气"args_schema:Type[BaseModel]=WeatherInputdef_run(self,city:str)->str:"""同步执行"""# 实际调用天气APIapi_key="YOUR_API_KEY"url=f"http://api.weatherapi.com/v1/current.json?key={api_key}&q={city}"response=requests.get(url)data=response.json()returnf"{city}天气：{data['current']['condition']['text']}，温度：{data['current']['temp_c']}°C"asyncdef_arun(self,city:str)->str:"""异步执行"""# 异步版本实现pass# 代码执行工具classCodeExecutor(BaseTool):"""Python代码执行工具"""name="python"description="执行Python代码并返回结果"def_run(self,code:str)->str:"""安全执行代码"""importsysfromioimportStringIO# 捕获输出old_stdout=sys.stdout sys.stdout=StringIO()try:# 限制执行环境exec(code,{"__builtins__":{"print":print,"len":len,"range":range,"str":str,"int":int,"float":float,}})output=sys.stdout.getvalue()exceptExceptionase:output=f"错误：{e}"finally:sys.stdout=old_stdoutreturnoutput2. 工具编排
classToolOrchestrator:"""工具编排器"""def__init__(self):self.tools={}defregister_tool(self,tool:BaseTool):"""注册工具"""self.tools[tool.name]=tooldefcreate_toolkit(self,task_type:str):"""根据任务类型选择工具集"""toolkits={"research":["search","wikipedia","arxiv"],"coding":["python","shell","file_ops"],"analysis":["calculator","pandas","plot"],"communication":["email","slack","calendar"]}selected_tools=[]fortool_nameintoolkits.get(task_type,[]):iftool_nameinself.tools:selected_tools.append(self.tools[tool_name])returnselected_toolsFunction Calling（OpenAI风格）
实现Function Calling
importjsonfromtypingimportList,Dict,AnyclassFunctionCallingAgent:"""支持Function Calling的Agent"""def__init__(self,model="gpt-3.5-turbo"):self.model=model self.functions=[]defregister_function(self,func,description:str,parameters:dict):"""注册函数"""self.functions.append({"name":func.__name__,"description":description,"parameters":parameters,"func":func})defrun(self,messages:List[Dict]):"""执行对话"""importopenai# 准备函数定义functions_def=[{"name":f["name"],"description":f["description"],"parameters":f["parameters"]}forfinself.functions]# 调用OpenAI APIresponse=openai.ChatCompletion.create(model=self.model,messages=messages,functions=functions_def,function_call="auto")message=response.choices[0].message# 检查是否需要调用函数ifmessage.get("function_call"):function_name=message["function_call"]["name"]function_args=json.loads(message["function_call"]["arguments"])# 执行函数func=next(f["func"]forfinself.functionsiff["name"]==function_name)result=func(**function_args)# 将结果返回给模型messages.append(message)messages.append({"role":"function","name":function_name,"content":str(result)})# 获取最终回答final_response=openai.ChatCompletion.create(model=self.model,messages=messages)returnfinal_response.choices[0].message["content"]returnmessage["content"]# 使用示例agent=FunctionCallingAgent()# 注册函数defget_weather(location:str,unit:str="celsius"):"""获取天气信息"""returnf"{location}的温度是25°{unit[0].upper()}"agent.register_function(get_weather,"获取指定地点的天气",{"type":"object","properties":{"location":{"type":"string","description":"城市名"},"unit":{"type":"string","enum":["celsius","fahrenheit"]}},"required":["location"]})# 执行result=agent.run([{"role":"user","content":"北京今天天气怎么样？"}])多Agent协作系统
1. Agent角色定义
classSpecializedAgent:"""专业Agent基类"""def__init__(self,name:str,role:str,skills:List[str]):self.name=name self.role=role self.skills=skills self.llm=ChatOpenAI(temperature=0.7)defcan_handle(self,task:str)->bool:"""判断是否能处理任务"""prompt=f""" 作为{self.role}，我的技能包括：{', '.join(self.skills)}判断我是否适合处理以下任务：{task}回答：是/否 """response=self.llm.predict(prompt)return"是"inresponsedefprocess(self,task:str)->str:"""处理任务"""prompt=f""" 角色：{self.role}技能：{', '.join(self.skills)}任务：{task}请完成任务： """returnself.llm.predict(prompt)# 创建专业Agentresearcher=SpecializedAgent("研究员","信息研究专家",["搜索","文献分析","数据收集"])coder=SpecializedAgent("程序员","软件开发专家",["Python","调试","代码优化"])analyst=SpecializedAgent("分析师","数据分析专家",["统计分析","可视化","报告撰写"])2. 协作框架
classMultiAgentSystem:"""多Agent协作系统"""def__init__(self):self.agents=[]self.coordinator=ChatOpenAI(temperature=0)self.conversation_history=[]defadd_agent(self,agent:SpecializedAgent):"""添加Agent"""self.agents.append(agent)defexecute(self,task:str):"""执行复杂任务"""# 1. 任务分解subtasks=self._decompose_task(task)results={}forsubtaskinsubtasks:# 2. 分配Agentagent=self._assign_agent(subtask)ifagent:# 3. 执行子任务result=agent.process(subtask)results[subtask]={"agent":agent.name,"result":result}# 4. 记录对话self.conversation_history.append({"agent":agent.name,"task":subtask,"result":result})# 5. 整合结果returnself._synthesize_results(task,results)def_decompose_task(self,task:str)->List[str]:"""任务分解"""prompt=f""" 将以下复杂任务分解为3-5个子任务： 任务：{task}输出格式： 1. 子任务1 2. 子任务2 ... """response=self.coordinator.predict(prompt)subtasks=[]forlineinresponse.split('\n'):ifline.strip()andline[0].isdigit():subtasks.append(line.split('.',1)[1].strip())returnsubtasksdef_assign_agent(self,subtask:str)->SpecializedAgent:"""分配最合适的Agent"""foragentinself.agents:ifagent.can_handle(subtask):returnagentreturnNonedef_synthesize_results(self,task:str,results:Dict):"""整合结果"""prompt=f""" 原始任务：{task}各Agent完成情况：{json.dumps(results,ensure_ascii=False,indent=2)}请整合以上结果，给出最终答案： """returnself.coordinator.predict(prompt)生产级Agent实战
完整的客服Agent系统
classCustomerServiceAgent:"""客服Agent系统"""def__init__(self):self.llm=ChatOpenAI(temperature=0.3)self.memory=ConversationBufferWindowMemory(k=5)self.knowledge_base=None# RAG系统self.tools=self._init_tools()def_init_tools(self):"""初始化工具集"""return[Tool(name="查询订单",func=self._query_order),Tool(name="退款处理",func=self._process_refund),Tool(name="人工转接",func=self._transfer_human),Tool(name="知识库",func=self._search_kb),]defhandle_customer(self,message:str,customer_id:str):"""处理客户消息"""# 1. 意图识别intent=self._identify_intent(message)# 2. 情绪检测emotion=self._detect_emotion(message)# 3. 获取客户历史history=self._get_customer_history(customer_id)# 4. 构建Agentagent=initialize_agent(self.tools,self.llm,agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,memory=self.memory,verbose=True)# 5. 生成回复context=f""" 客户ID:{customer_id}意图:{intent}情绪:{emotion}历史问题:{history[-3:] if history else '无'}客户消息:{message}"""response=agent.run(context)# 6. 记录交互self._log_interaction(customer_id,message,response)returnresponsedef_identify_intent(self,message:str)->str:"""意图识别"""intents=["查询","投诉","退款","咨询","其他"]prompt=f""" 识别客户意图： 消息：{message}可选意图：{', '.join(intents)}输出意图： """returnself.llm.predict(prompt).strip()def_detect_emotion(self,message:str)->str:"""情绪检测"""prompt=f""" 检测客户情绪（积极/中性/消极/愤怒）： 消息：{message}情绪： """returnself.llm.predict(prompt).strip()调试与优化
classAgentDebugger:"""Agent调试工具"""def__init__(self,agent):self.agent=agent self.traces=[]deftrace_execution(self,task:str):"""跟踪执行过程"""importtime trace={"task":task,"steps":[],"start_time":time.time()}# 劫持Agent的执行original_run=self.agent.rundeftraced_run(*args,**kwargs):step_start=time.time()result=original_run(*args,**kwargs)trace["steps"].append({"input":args[0]ifargselsekwargs,"output":result,"duration":time.time()-step_start})returnresult self.agent.run=traced_run result=self.agent.run(task)self.agent.run=original_run trace["end_time"]=time.time()trace["total_duration"]=trace["end_time"]-trace["start_time"]trace["result"]=result self.traces.append(trace)returntracedefanalyze_performance(self):"""性能分析"""ifnotself.traces:return"无执行记录"total_time=sum(t["total_duration"]fortinself.traces)avg_time=total_time/len(self.traces)# 找出最慢的步骤slowest_steps=[]fortraceinself.traces:forstepintrace["steps"]:ifstep["duration"]>avg_time*0.5:slowest_steps.append({"task":trace["task"],"step":step["input"],"duration":step["duration"]})return{"总执行次数":len(self.traces),"平均耗时":f"{avg_time:.2f}秒","最慢步骤":slowest_steps[:5]}
