刘涛做的网站网站地图写法

刘涛做的网站,网站地图写法,自己做一个小程序需要多少钱,青岛网页设计 学校【LLM实操系列07】Agent开发#xff1a;构建自主AI智能体 在开始之前#xff0c;建议先完成第04篇#xff08;理解ReAct概念#xff09;和第03篇#xff08;API调用#xff09;。你需要理解工具调用和思考-行动-观察循环的基本概念#xff0c;并安装langchain及相关工具…【LLM实操系列07】Agent开发构建自主AI智能体在开始之前建议先完成第04篇理解ReAct概念和第03篇API调用。你需要理解工具调用和思考-行动-观察循环的基本概念并安装langchain及相关工具库。10分钟实现第一个Agent什么是AgentAgent LLM 工具 记忆 规划最简Agent实现# pip install langchain openai wikipedia duckduckgo-searchfromlangchain.agentsimportinitialize_agent,Toolfromlangchain.agentsimportAgentTypefromlangchain.chat_modelsimportChatOpenAIfromlangchain.toolsimportDuckDuckGoSearchRun# 1. 定义工具searchDuckDuckGoSearchRun()tools[Tool(nameSearch,funcsearch.run,description搜索引擎用于查找实时信息)]# 2. 创建AgentllmChatOpenAI(temperature0)agentinitialize_agent(tools,llm,agentAgentType.ZERO_SHOT_REACT_DESCRIPTION,verboseTrue)# 3. 执行任务resultagent.run(今天上海的天气如何)print(result)ReAct框架思考→行动→观察基于第04篇的ReAct完整实现# 注第04篇介绍了ReAct的概念这里实现完整的工具调用classReActAgent:ReAct模式Agent实现完整版def__init__(self,llm,tools):self.llmllm self.tools{tool.name:toolfortoolintools}self.max_iterations5defrun(self,task:str):执行任务promptf 任务{task}可用工具{self._format_tools()}使用以下格式 Thought: 我需要思考下一步做什么 Action: 工具名称 Action Input: 工具输入参数 Observation: 工具返回结果 ... (重复N次) Thought: 我知道最终答案了 Final Answer: 最终答案 开始 foriinrange(self.max_iterations):# 获取LLM输出responseself.llm.predict(prompt)# 解析行动actionself._parse_action(response)ifaction[type]Final Answer:returnaction[content]# 执行工具toolself.tools[action[tool]]observationtool.func(action[input])# 添加观察结果promptf Thought:{action[thought]}Action:{action[tool]}Action Input:{action[input]}Observation:{observation}return达到最大迭代次数def_parse_action(self,text:str):解析LLM输出ifFinal Answer:intext:return{type:Final Answer,content:text.split(Final Answer:)[-1].strip()}# 提取Action和Inputlinestext.strip().split(\n)thoughtactionaction_inputforlineinlines:ifline.startswith(Thought:):thoughtline[8:].strip()elifline.startswith(Action:):actionline[7:].strip()elifline.startswith(Action Input:):action_inputline[13:].strip()return{type:Action,thought:thought,tool:action,input:action_input}工具开发与集成1. 自定义工具fromtypingimportOptional,TypefrompydanticimportBaseModel,Fieldfromlangchain.toolsimportBaseToolimportrequestsclassWeatherInput(BaseModel):天气查询输入city:strField(description城市名称)classWeatherTool(BaseTool):天气查询工具nameweatherdescription查询指定城市的天气args_schema:Type[BaseModel]WeatherInputdef_run(self,city:str)-str:同步执行# 实际调用天气APIapi_keyYOUR_API_KEYurlfhttp://api.weatherapi.com/v1/current.json?key{api_key}q{city}responserequests.get(url)dataresponse.json()returnf{city}天气{data[current][condition][text]}温度{data[current][temp_c]}°Casyncdef_arun(self,city:str)-str:异步执行# 异步版本实现pass# 代码执行工具classCodeExecutor(BaseTool):Python代码执行工具namepythondescription执行Python代码并返回结果def_run(self,code:str)-str:安全执行代码importsysfromioimportStringIO# 捕获输出old_stdoutsys.stdout sys.stdoutStringIO()try:# 限制执行环境exec(code,{__builtins__:{print:print,len:len,range:range,str:str,int:int,float:float,}})outputsys.stdout.getvalue()exceptExceptionase:outputf错误{e}finally:sys.stdoutold_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 CallingOpenAI风格实现Function CallingimportjsonfromtypingimportList,Dict,AnyclassFunctionCallingAgent:支持Function Calling的Agentdef__init__(self,modelgpt-3.5-turbo):self.modelmodel 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 APIresponseopenai.ChatCompletion.create(modelself.model,messagesmessages,functionsfunctions_def,function_callauto)messageresponse.choices[0].message# 检查是否需要调用函数ifmessage.get(function_call):function_namemessage[function_call][name]function_argsjson.loads(message[function_call][arguments])# 执行函数funcnext(f[func]forfinself.functionsiff[name]function_name)resultfunc(**function_args)# 将结果返回给模型messages.append(message)messages.append({role:function,name:function_name,content:str(result)})# 获取最终回答final_responseopenai.ChatCompletion.create(modelself.model,messagesmessages)returnfinal_response.choices[0].message[content]returnmessage[content]# 使用示例agentFunctionCallingAgent()# 注册函数defget_weather(location:str,unit:strcelsius):获取天气信息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]})# 执行resultagent.run([{role:user,content:北京今天天气怎么样}])多Agent协作系统1. Agent角色定义classSpecializedAgent:专业Agent基类def__init__(self,name:str,role:str,skills:List[str]):self.namename self.rolerole self.skillsskills self.llmChatOpenAI(temperature0.7)defcan_handle(self,task:str)-bool:判断是否能处理任务promptf 作为{self.role}我的技能包括{, .join(self.skills)}判断我是否适合处理以下任务{task}回答是/否 responseself.llm.predict(prompt)return是inresponsedefprocess(self,task:str)-str:处理任务promptf 角色{self.role}技能{, .join(self.skills)}任务{task}请完成任务 returnself.llm.predict(prompt)# 创建专业AgentresearcherSpecializedAgent(研究员,信息研究专家,[搜索,文献分析,数据收集])coderSpecializedAgent(程序员,软件开发专家,[Python,调试,代码优化])analystSpecializedAgent(分析师,数据分析专家,[统计分析,可视化,报告撰写])2. 协作框架classMultiAgentSystem:多Agent协作系统def__init__(self):self.agents[]self.coordinatorChatOpenAI(temperature0)self.conversation_history[]defadd_agent(self,agent:SpecializedAgent):添加Agentself.agents.append(agent)defexecute(self,task:str):执行复杂任务# 1. 任务分解subtasksself._decompose_task(task)results{}forsubtaskinsubtasks:# 2. 分配Agentagentself._assign_agent(subtask)ifagent:# 3. 执行子任务resultagent.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]:任务分解promptf 将以下复杂任务分解为3-5个子任务 任务{task}输出格式 1. 子任务1 2. 子任务2 ... responseself.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:分配最合适的Agentforagentinself.agents:ifagent.can_handle(subtask):returnagentreturnNonedef_synthesize_results(self,task:str,results:Dict):整合结果promptf 原始任务{task}各Agent完成情况{json.dumps(results,ensure_asciiFalse,indent2)}请整合以上结果给出最终答案 returnself.coordinator.predict(prompt)生产级Agent实战完整的客服Agent系统classCustomerServiceAgent:客服Agent系统def__init__(self):self.llmChatOpenAI(temperature0.3)self.memoryConversationBufferWindowMemory(k5)self.knowledge_baseNone# RAG系统self.toolsself._init_tools()def_init_tools(self):初始化工具集return[Tool(name查询订单,funcself._query_order),Tool(name退款处理,funcself._process_refund),Tool(name人工转接,funcself._transfer_human),Tool(name知识库,funcself._search_kb),]defhandle_customer(self,message:str,customer_id:str):处理客户消息# 1. 意图识别intentself._identify_intent(message)# 2. 情绪检测emotionself._detect_emotion(message)# 3. 获取客户历史historyself._get_customer_history(customer_id)# 4. 构建Agentagentinitialize_agent(self.tools,self.llm,agentAgentType.CONVERSATIONAL_REACT_DESCRIPTION,memoryself.memory,verboseTrue)# 5. 生成回复contextf 客户ID:{customer_id}意图:{intent}情绪:{emotion}历史问题:{history[-3:] if history else 无}客户消息:{message}responseagent.run(context)# 6. 记录交互self._log_interaction(customer_id,message,response)returnresponsedef_identify_intent(self,message:str)-str:意图识别intents[查询,投诉,退款,咨询,其他]promptf 识别客户意图 消息{message}可选意图{, .join(intents)}输出意图 returnself.llm.predict(prompt).strip()def_detect_emotion(self,message:str)-str:情绪检测promptf 检测客户情绪积极/中性/消极/愤怒 消息{message}情绪 returnself.llm.predict(prompt).strip()调试与优化classAgentDebugger:Agent调试工具def__init__(self,agent):self.agentagent self.traces[]deftrace_execution(self,task:str):跟踪执行过程importtime trace{task:task,steps:[],start_time:time.time()}# 劫持Agent的执行original_runself.agent.rundeftraced_run(*args,**kwargs):step_starttime.time()resultoriginal_run(*args,**kwargs)trace[steps].append({input:args[0]ifargselsekwargs,output:result,duration:time.time()-step_start})returnresult self.agent.runtraced_run resultself.agent.run(task)self.agent.runoriginal_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_timesum(t[total_duration]fortinself.traces)avg_timetotal_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]}