生产级(≈500 行结构思路)的 Agent 系统模板

📅 发布时间:2026/7/8 18:56:42 👁️ 浏览次数:
生产级(≈500 行结构思路)的 Agent 系统模板
接近真实系统LangGraphworkflow runtimeMicrosoft AutoGenmulti-agent communicationSWE-agenttool environment agent这个版本包含PlannerTool RouterRAG MemoryReflectionAgent RuntimeEnvironmentTask Graph重点不是行数而是架构完整度。1 真实项目目录结构真实项目一般长这样agent_system/ │ ├── agent_runtime.py ├── planner.py ├── memory.py ├── tools.py ├── reflection.py ├── rag.py ├── environment.py ├── workflow.py ├── llm.py └── main.py这样代码就会接近500行左右的 MVP agent framework。2 LLM Wrapper统一管理 LLM 调用。# llm.pyimportopenaidefchat(prompt):responseopenai.ChatCompletion.create(modelgpt-4o-mini,messages[{role:user,content:prompt}])returnresponse.choices[0].message[content]生产环境通常会加retrystreamingcost tracking3 Planner任务规划# planner.pyimportjsonfromllmimportchatdefplan(goal,context):promptf You are an AI planner. Goal:{goal}Context:{context}Break the goal into steps. Return JSON list. resultchat(prompt)try:stepsjson.loads(result)except:steps[goal]returnsteps输出示例[search population of tokyo,calculate double]4 Tool System# tools.pydefsearch(query):returnfsearch results for{query}defcalculator(expr):try:returnstr(eval(expr))except:returnerrorTOOLS{search:search,calculator:calculator}5 Tool Router# tools.pyfromllmimportchatdefselect_tool(task):promptf Select best tool. task:{task}tools: search calculator Return tool name or NONE toolchat(prompt).strip()iftoolinTOOLS:returntoolreturnNone6 Memory System# memory.pyclassMemory:def__init__(self):self.history[]defadd(self,item):self.history.append(item)defcontext(self):return\n.join(self.history[-20:])生产系统通常会加vector databasesemantic search7 RAG Retrieval# rag.pydefretrieve(query):docs[Tokyo population is about 14 million]returndocs真实项目通常接ChromaPinecone8 Reflection System# reflection.pyfromllmimportchatdefevaluate(task,result):promptf Task:{task}Result:{result}Is result correct? Reply YES or NO decisionchat(prompt)returnYESindecision这种结构和SWE-agent 的critic agent很类似。9 EnvironmentAgent 与系统交互。# environment.pyimportsubprocessdefrun_shell(command):try:resultsubprocess.check_output(command,shellTrue)returnresult.decode()except:returnshell error这种环境模式类似OpenDevin10 Workflow Engine简单任务 DAG。# workflow.pyclassWorkflow:def__init__(self):self.tasks[]defadd(self,task):self.tasks.append(task)defrun(self,executor):fortaskinself.tasks:executor(task)复杂版本会变成graph execution state machine类似LangGraph。11 Agent Runtime核心循环# agent_runtime.pyfromplannerimportplanfromtoolsimportselect_tool,TOOLSfromreflectionimportevaluatefrommemoryimportMemoryfromragimportretrievefromllmimportchatclassAgent:def__init__(self):self.memoryMemory()defexecute(self,task):toolselect_tool(task)iftool:resultTOOLS[tool](task)else:resultchat(task)returnresultdefrun(self,goal):contextself.memory.context()docsretrieve(goal)context\n.join(docs)tasksplan(goal,context)fortaskintasks:print(TASK:,task)resultself.execute(task)print(RESULT:,result)okevaluate(task,result)ifnotok:print(retrying)resultself.execute(task)self.memory.add(task)self.memory.add(result)12 Main# main.pyfromagent_runtimeimportAgent agentAgent()agent.run(find population of tokyo and double it)完整 Agent 运行流程User Goal ↓ Planner ↓ Task List ↓ Tool Router ↓ Tool Execution ↓ Reflection ↓ Memory Update循环Observe → Plan → Act → Reflect真实生产系统还会增加1 Workflow Graph类似LangGraphplanner ↓ research ↓ analysis ↓ writer2 Multi-Agent类似Microsoft AutoGenPlanner Agent Coder Agent Reviewer Agent3 Computer Environment类似OpenDevinAgent 可以edit files run tests execute commands一句话总结 Production Agent真实系统基本是Agent Runtime Planner Tools Memory (RAG) Reflection Environment