Agent 项目落地模板

Agent 项目落地模板1. 路线选择原则如果你要真的开工我建议默认采用这个路线先做 L1/L3不要一开始做纯 L2 loop agent目录结构按“可升级到 graph”来设计工具、状态、规划、执行器要分开所有 side-effect tool 都必须可审计一句话先把 Agent 当成“LLM 驱动的 workflow 系统”来做而不是“全自动智能体”来做。2. 模版适合项目类型这份模板适合以下项目企业内部知识助手Research Agent自动报告生成多工具办公自动化运维/数据分析助手面向 API 的 Agent 服务不太适合纯聊天机器人只有单次 function calling 的极简服务极端高频低延迟系统3. 项目总体架构先看整体分层。client / ui ↓ api layer ↓ application layer ↓ agent runtime ├── planner ├── executor ├── tools ├── state ├── memory ├── policies └── observability ↓ infra layer ├── llm provider ├── cache ├── db ├── queue └── tracing/logging更具体一点User Request ↓ FastAPI Router ↓ Agent Service ↓ Task Interpreter ↓ Planner ↓ Plan Validator ↓ Executor ├── Tool Registry ├── State Store ├── Policy Check ├── Cache └── Retry Manager ↓ Final Synthesizer ↓ Response4. 目录结构这是一个比较实用的目录设计。agent_app/ ├── app/ │ ├── api/ │ │ ├── routers/ │ │ │ ├── health.py │ │ │ ├── chat.py │ │ │ └── tasks.py │ │ └── deps.py │ │ │ ├── core/ │ │ ├── config.py │ │ ├── logging.py │ │ ├── exceptions.py │ │ └── security.py │ │ │ ├── domain/ │ │ ├── models/ │ │ │ ├── task.py │ │ │ ├── plan.py │ │ │ ├── state.py │ │ │ ├── tool.py │ │ │ └── response.py │ │ ├── enums.py │ │ └── policies.py │ │ │ ├── application/ │ │ ├── services/ │ │ │ ├── agent_service.py │ │ │ ├── planning_service.py │ │ │ ├── execution_service.py │ │ │ └── evaluation_service.py │ │ └── usecases/ │ │ ├── run_agent.py │ │ └── run_workflow.py │ │ │ ├── agent/ │ │ ├── planner/ │ │ │ ├── base.py │ │ │ ├── llm_planner.py │ │ │ └── prompts.py │ │ ├── executor/ │ │ │ ├── base.py │ │ │ ├── sequential_executor.py │ │ │ ├── graph_executor.py │ │ │ └── node_runner.py │ │ ├── tools/ │ │ │ ├── base.py │ │ │ ├── registry.py │ │ │ ├── web_search.py │ │ │ ├── database_query.py │ │ │ ├── python_exec.py │ │ │ └── email_sender.py │ │ ├── memory/ │ │ │ ├── base.py │ │ │ ├── short_term.py │ │ │ └── vector_memory.py │ │ ├── state/ │ │ │ ├── store.py │ │ │ └── serializers.py │ │ ├── policies/ │ │ │ ├── tool_policy.py │ │ │ ├── budget_policy.py │ │ │ └── safety_policy.py │ │ ├── prompts/ │ │ │ ├── system_prompts.py │ │ │ ├── planner_prompts.py │ │ │ └── synthesis_prompts.py │ │ └── runtime/ │ │ ├── orchestrator.py │ │ ├── guards.py │ │ └── context_builder.py │ │ │ ├── integrations/ │ │ ├── llm/ │ │ │ ├── base.py │ │ │ ├── openai_client.py │ │ │ ├── anthropic_client.py │ │ │ └── gemini_client.py │ │ ├── storage/ │ │ │ ├── redis_store.py │ │ │ ├── postgres_store.py │ │ │ └── s3_store.py │ │ ├── cache/ │ │ │ └── redis_cache.py │ │ └── tracing/ │ │ ├── logger.py │ │ ├── metrics.py │ │ └── opentelemetry.py │ │ │ ├── tests/ │ │ ├── unit/ │ │ ├── integration/ │ │ ├── evals/ │ │ └── fixtures/ │ │ │ └── main.py │ ├── scripts/ │ ├── run_local.sh │ ├── seed_tools.py │ └── eval_runner.py │ ├── configs/ │ ├── dev.yaml │ ├── prod.yaml │ └── prompts.yaml │ ├── Dockerfile ├── docker-compose.yml ├── pyproject.toml └── README.md5. 各模块职责说明5.1 api/负责对外 HTTP 接口。应包含/chat/tasks/run/health/metrics可选原则API 层不要写 Agent 核心逻辑只做参数解析、鉴权、调用 usecase5.2 domain/这里放核心数据结构不要混业务逻辑。核心模型建议task.pyfrompydanticimportBaseModelfromtypingimportOptional,Dict,AnyclassTaskRequest(BaseModel):user_id:strquery:strmetadata:Dict[str,Any]{}session_id:Optional[str]Noneplan.pyfrompydanticimportBaseModelfromtypingimportList,Dict,Any,OptionalclassPlanStep(BaseModel):name:strkind:str# tool | llmtool_name:Optional[str]Noneinstruction:Optional[str]Noneargs:Dict[str,Any]{}depends_on:List[str][]classExecutionPlan(BaseModel):steps:List[PlanStep]goal:strstate.pyfrompydanticimportBaseModel,FieldfromtypingimportDict,AnyclassAgentState(BaseModel):values:Dict[str,Any]Field(default_factorydict)defset(self,key:str,value:Any):self.values[key]valuedefget(self,key:str,defaultNone):returnself.values.get(key,default)tool.pyfrompydanticimportBaseModelfromtypingimportDict,AnyclassToolCall(BaseModel):tool_name:strargs:Dict[str,Any]classToolResult(BaseModel):success:booloutput:Any error:str|NoneNone5.3 application/这里是用例编排层把 API 请求转成领域服务调用。例子run_agent.pyrun_workflow.pyrun_agent.pyclassRunAgentUseCase:def__init__(self,agent_service):self.agent_serviceagent_serviceasyncdefexecute(self,request):returnawaitself.agent_service.run(request)5.4 agent/planner/负责任务规划。职责解析用户目标输出线性 planL3或 task graphL4不做真实执行规划器接口fromabcimportABC,abstractmethodclassBasePlanner(ABC):abstractmethodasyncdefcreate_plan(self,task_request):...一个简单 plannerclassLLMPlanner(BasePlanner):def__init__(self,llm_client,prompt_builder):self.llmllm_client self.prompt_builderprompt_builderasyncdefcreate_plan(self,task_request):promptself.prompt_builder.build(task_request)rawawaitself.llm.generate(prompt)returnparse_plan(raw)重要原则planner 输出必须结构化最好 JSON schema 校验planner 不要直接执行工具5.5 agent/executor/负责执行计划是系统的关键。L3顺序执行器classSequentialExecutor:def__init__(self,tool_registry,llm_client,state_store):self.tool_registrytool_registry self.llmllm_client self.state_storestate_storeasyncdefexecute(self,plan,state):forstepinplan.steps:ifstep.kindtool:toolself.tool_registry.get(step.tool_name)resultawaittool.run(step.args,state)state.set(step.name,result)elifstep.kindllm:promptbuild_step_prompt(step,state)resultawaitself.llm.generate(prompt)state.set(step.name,result)returnstateL4图执行器classGraphExecutor:def__init__(self,node_runner):self.node_runnernode_runnerasyncdefexecute(self,plan,state):completedset()whilelen(completed)len(plan.steps):ready[sforsinplan.stepsifs.namenotincompletedandall(depincompletedfordepins.depends_on)]resultsawaitrun_parallel([self.node_runner.run(step,state)forstepinready])forstep,resultinzip(ready,results):state.set(step.name,result)completed.add(step.name)returnstate5.6 agent/tools/这里是最容易做烂的地方。建议每个工具都遵循统一接口。工具基类fromabcimportABC,abstractmethodclassBaseTool(ABC):name:strdescription:strabstractmethodasyncdefrun(self,args,state):...示例网页搜索classWebSearchTool(BaseTool):nameweb_searchdescriptionSearch the web for informationasyncdefrun(self,args,state):queryargs[query]# 调第三方搜索 APIreturn{query:query,results:[result1,result2]}工具注册表classToolRegistry:def__init__(self):self._tools{}defregister(self,tool):self._tools[tool.name]tooldefget(self,name):ifnamenotinself._tools:raiseValueError(fTool not found:{name})returnself._tools[name]工具设计原则每个工具都要明确输入 schema输出 schema是否有副作用权限级别超时重试策略5.7 agent/policies/这是生产系统必须有的。推荐至少有三类策略1工具权限策略classToolPolicy:defis_allowed(self,user,tool_name):forbidden{delete_user,refund_payment}returntool_namenotinforbidden2预算策略classBudgetPolicy:def__init__(self,max_steps8,max_tokens20000):self.max_stepsmax_steps self.max_tokensmax_tokens3安全策略prompt injection 检测PII 屏蔽敏感动作审批外发邮件审核5.8 agent/runtime/这是 orchestration 的核心。orchestrator.pyclassAgentOrchestrator:def__init__(self,planner,executor,synthesizer,policies):self.plannerplanner self.executorexecutor self.synthesizersynthesizer self.policiespoliciesasyncdefrun(self,task_request):self.policies.check_request(task_request)planawaitself.planner.create_plan(task_request)self.policies.check_plan(plan)stateAgentState()stateawaitself.executor.execute(plan,state)returnawaitself.synthesizer.generate(task_request,state)guards.py负责max stepstimeoutloop detectiontool retry guard5.9 integrations/llm/这一层要做成可替换。接口classBaseLLMClient:asyncdefgenerate(self,prompt:str)-str:raiseNotImplementedErrorOpenAI 示例classOpenAIClient(BaseLLMClient):def__init__(self,client,model):self.clientclient self.modelmodelasyncdefgenerate(self,prompt:str)-str:respawaitself.client.responses.create(modelself.model,inputprompt)returnresp.output_text建议把以下能力封装掉重试timeouttoken 统计tracingtool call 抽象5.10 integrations/tracing/生产上必须有 tracing。至少记录request_idsession_iduser_idplantool_nametool_argstool_result summarytoken usagelatencyfinal answer quality signal6. 一个最小可运行的 L3 Agent 模板下面给你一个最小骨架。6.1agent_service.pyclassAgentService:def__init__(self,orchestrator):self.orchestratororchestratorasyncdefrun(self,request):returnawaitself.orchestrator.run(request)6.2llm_planner.pyclassLLMPlanner:def__init__(self,llm_client):self.llmllm_clientasyncdefcreate_plan(self,task_request):# 实际项目里这里应该要求输出 JSON# 这里简化成固定计划returnExecutionPlan(goaltask_request.query,steps[PlanStep(namesearch,kindtool,tool_nameweb_search,args{query:task_request.query}),PlanStep(namesummary,kindllm,instructionSummarize the search results)])6.3sequential_executor.pyclassSequentialExecutor:def__init__(self,tool_registry,llm_client):self.tool_registrytool_registry self.llmllm_clientasyncdefexecute(self,plan,state):forstepinplan.steps:ifstep.kindtool:toolself.tool_registry.get(step.tool_name)resultawaittool.run(step.args,state)state.set(step.name,result)elifstep.kindllm:promptf{step.instruction}\nContext:\n{state.values}resultawaitself.llm.generate(prompt)state.set(step.name,result)returnstate6.4web_search.pyclassWebSearchTool:nameweb_searchdescriptionSearch web contentasyncdefrun(self,args,state):queryargs[query]# 假装调用搜索引擎return{query:query,results:[{title:Result A,snippet:Info A},{title:Result B,snippet:Info B},]}6.5synthesizer.pyclassFinalSynthesizer:def__init__(self,llm_client):self.llmllm_clientasyncdefgenerate(self,task_request,state):promptf User request:{task_request.query}Execution state:{state.values}Produce a final user-friendly answer. returnawaitself.llm.generate(prompt)6.6main.pyfromfastapiimportFastAPI appFastAPI()app.post(/tasks/run)asyncdefrun_task(payload:dict):requestTaskRequest(**payload)resultawaitagent_service.run(request)return{result:result}7. 从 L3 升级到 L4 的设计预留如果你现在做 L3但以后可能升级到 L4建议现在就预留好这些接口。7.1 PlanStep 中预留依赖字段即使暂时不用 graph也先加上depends_on:List[str][]这样以后可以平滑升级 DAG。7.2 Executor 接口抽象化classBaseExecutor(ABC):abstractmethodasyncdefexecute(self,plan,state):...这样你可以轻松切换SequentialExecutorGraphExecutor7.3 State 不要写死成字符串拼接要用结构化 statestate.set(search_result,{...})state.set(analysis,{...})这样 node 之间传递才可靠。7.4 Tool 输出尽量结构化不要只返回一坨文本。坏例子returnI found some web pages...好例子return{results:[...],source_count:5,query:query}8. 推荐技术栈按“务实落地”来推荐。8.1 Web/APIFastAPI8.2 数据模型Pydantic8.3 LLM 接入官方 SDKOpenAI / Anthropic / Gemini或 LiteLLM 做多模型适配8.4 缓存Redis8.5 状态/持久化Postgres或 MongoDB如果偏文档8.6 可观测性OpenTelemetryPrometheus GrafanaLangfuse / Helicone / Phoenix二选一类8.7 任务队列Celery / Dramatiq / Arq批处理可用 Prefect / Airflow8.8 向量检索pgvector / Qdrant / Weaviate9. 推荐优先开源项目按你做的层级来选。9.1 如果你做 L1/L3优先官方 SDKPydanticAILiteLLM自研 planner executor这是我最推荐的组合最稳。9.2 如果你做 L4优先LangGraphPrefect / Dagster / Airflow偏工作流自研 graph executor如果你有平台团队9.3 如果你做实验/原型可用LangChainAutoGenCrewAI但建议用来验证想法不要默认当最终生产内核10. 测试策略模板Agent 项目最容易忽略测试。建议至少分三层。10.1 单元测试测planner 输出结构tool 输入输出policy 校验state 更新逻辑10.2 集成测试测API → planner → executor → tool → synthesizer 全链路第三方 API mock超时/失败重试10.3 evals这是 Agent 项目的核心测试。准备一个数据集[{query:分析 AI coding 市场趋势,expected_signals:[market,trend,competitor]},{query:查询订单123状态,expected_tool:order_query}]评估是否选对工具是否成功完成任务latencytoken costanswer quality11. 生产环境必须加的保护这个很关键。11.1 Max limitsmax_stepsmax_tool_callsmax_tokensmax_execution_time11.2 Tool allowlist按用户角色控制工具allowed_tools{guest:[web_search],employee:[web_search,db_query],admin:[web_search,db_query,send_email]}11.3 Side-effect tools 审批例如send_emailcreate_ticketupdate_db应支持dry-runhuman approvalaudit log11.4 Fallback当 planner 失败时fallback 到固定 workflowfallback 到简单回答fallback 到人工12. 一个简化版 README 模板你项目里可以这么写。Project OverviewThis project implements a production-oriented AI Agent system with:planner executor architecturestructured toolssafety policiesobservability and evaluation hooksArchitectureL3 by defaultL4-ready abstractionstool registrystate storepolicy engineRun locallycp.env.example .envdocker-composeup-duvicorn app.main:app--reloadTestpytest python scripts/eval_runner.py13. 最推荐的最小落地方案如果你现在就想开工我建议你先做这个MVP 架构第一版FastAPIPydantic官方 LLM SDKToolRegistryLLMPlannerSequentialExecutorRedis cachePostgres logsLangfuse tracing不要先做的不要先做 multi-agent不要先做复杂 memory不要先做自由 loop agent不要先做全图调度不要先做太重的框架绑定14. 最后给你一个“可直接抄”的精简版骨架app/ ├── api/ ├── domain/ ├── application/ ├── agent/ │ ├── planner/ │ ├── executor/ │ ├── tools/ │ ├── policies/ │ ├── runtime/ │ └── prompts/ ├── integrations/ │ ├── llm/ │ ├── cache/ │ ├── storage/ │ └── tracing/ └── main.py其中最核心的 5 个类是LLMPlannerSequentialExecutorToolRegistryAgentStateAgentOrchestrator先把这 5 个类做好你的项目就已经是一个真正可落地的 Agent 系统雏形了。如果你愿意我下一步可以直接继续给你两份非常实战的内容之一一份完整的 Python 代码骨架含类定义一份“Research Agent”示例项目模板把上面的目录直接填成可运行示例