FastAPI+LangGraph实战：从零构建可执行任务的编码助手-北京尧图网络

1. 为什么我要从零造一个编码助手市面上不缺代码补全工具也不缺能陪你聊技术栈的对话机器人。但真正落到日常开发里你会发现一个尴尬的断层聊天窗口能告诉你“这个报错通常是因为依赖版本冲突”可它不会帮你打开终端、定位到具体文件、改完代码再跑一遍测试。你依然要手动完成从“知道答案”到“验证答案”的全部动作。羲和XiheAgent要解决的就是这个断层。它的定位很明确——从代码问答走向任务执行。简单说它不只是回答“这段代码什么意思”而是能接收一个开发任务自己规划步骤、调用工具、读写文件、执行命令最后把结果反馈给你。适合谁参考如果你正在用 FastAPI 搭后端服务同时想用 LangGraph 把 LLM 的推理能力编排成可落地的执行流那这套设计思路可以直接抄作业。我选 FastAPI 做服务层核心原因是它的异步支持和依赖注入机制天然适合承载 Agent 这种“多轮调用外部工具”的场景。而 LangGraph 负责的是状态机式的流程编排——它和 LangChain 的区别在于LangChain 更像一条链式管道适合线性问答LangGraph 允许你定义节点和边支持循环、条件分支和状态持久化这才是任务执行类 Agent 真正需要的能力。羲和的核心就是一张 LangGraph 状态图每个节点承担一个明确的执行角色。2. 整体架构设计与技术选型拆解2.1 为什么是 FastAPI 加 LangGraph 这套组合先说 FastAPI。一个编码助手对外要暴露接口对内要管理会话状态、调用 LLM、执行本地命令。FastAPI 的async/await让这些 IO 密集型操作不会互相阻塞Pydantic 模型则保证了请求和响应结构的强校验。我实测下来用 FastAPI 做 Agent 服务层单机并发处理十几个会话请求基本没有压力。再说 LangGraph。很多人问 LangGraph 和 LangChain 到底什么区别我用一句话概括LangChain 的 Chain 是“一条路走到黑”LangGraph 的 Graph 是“可以掉头、可以绕圈、可以在路口做判断”。编码任务天然需要循环——比如 Agent 生成代码后执行失败它得回到“修改代码”节点重新来一遍。这种带环的执行逻辑用 LangGraph 的StateGraph表达最自然。提示如果你之前只用过 LangChain 的AgentExecutor迁移到 LangGraph 时最大的思维转变是——不要想着“一次调用返回结果”而是设计“状态如何在节点间流转”。2.2 项目目录结构怎么组织才不乱FastAPI 项目最容易犯的错是把所有路由、模型、业务逻辑堆在main.py里。羲和采用分层结构我直接给出经过实战验证的目录布局xihe-agent/ ├── app/ │ ├── main.py # FastAPI 入口注册路由和生命周期 │ ├── api/ │ │ ├── routes.py # 接口路由定义 │ │ └── schemas.py # Pydantic 请求/响应模型 │ ├── agent/ │ │ ├── graph.py # LangGraph 状态图定义 │ │ ├── nodes.py # 各执行节点的具体实现 │ │ ├── state.py # Agent 状态数据结构 │ │ └── tools.py # 工具函数文件读写、命令执行等 │ ├── core/ │ │ ├── config.py # 配置管理 │ │ └── llm.py # LLM 客户端初始化 │ └── services/ │ └── session.py # 会话管理与状态持久化 ├── tests/ ├── requirements.txt └── .env这个结构的好处是agent/目录完全独立于 Web 框架你可以单独测试状态图逻辑不用启动 FastAPI 服务。services/session.py负责把 LangGraph 的状态快照存下来这样即使服务重启用户的会话上下文也不会丢。2.3 Agent 状态设计整个系统的骨架LangGraph 的核心是状态。羲和的AgentState定义如下from typing import TypedDict, Annotated, Sequence from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[Sequence, add_messages] task: str # 当前任务描述 plan: list[str] # 规划出的步骤列表 current_step: int # 当前执行到第几步 code_context: dict # 相关代码文件内容 execution_result: str # 上一步执行结果 error_count: int # 连续错误计数 status: str # running / success / failed这里add_messages是 LangGraph 提供的 reducer它保证每次节点返回的新消息是追加而不是覆盖。error_count字段很关键——当 Agent 连续修改代码多次仍然执行失败时需要有一个熔断机制避免无限循环烧 token。3. 核心节点实现与实操要点3.1 规划节点把模糊任务拆成可执行步骤规划节点是整个 Agent 的“大脑”。用户输入“帮我给这个 FastAPI 项目加一个用户注册接口”规划节点要输出类似这样的步骤列表读取现有路由文件和用户模型定义生成注册接口的 Pydantic schema编写路由处理函数在 main.py 中注册路由运行测试验证实现上我用一个结构化输出的 prompt 让 LLM 返回 JSON 格式的步骤列表。关键技巧是在 prompt 中注入项目目录树否则 LLM 会凭空猜测文件路径。PLANNER_PROMPT 你是一个编码任务规划器。根据用户任务和项目结构输出一个 JSON 数组每个元素是一个可执行的步骤描述。项目结构 {project_tree} 用户任务{task} 要求 - 每个步骤必须对应一个明确的文件操作或命令执行 - 步骤数量控制在 3-8 个 - 只输出 JSON不要其他内容注意规划节点的输出一定要做 JSON 解析容错。LLM 有时会在 JSON 外面包一层 markdown 代码块标记我通常用正则先提取[...]部分再解析。3.2 执行节点工具调用与代码修改执行节点接收规划结果中的当前步骤决定调用哪个工具。羲和内置了四类工具工具名功能关键参数read_file读取文件内容path, start_line, end_linewrite_file写入或修改文件path, content, moderun_command执行 shell 命令command, timeoutsearch_code在项目中搜索关键词keyword, file_pattern执行节点的逻辑是把当前步骤描述和可用工具列表一起发给 LLM让 LLM 返回工具名和参数。这里有个坑——不要让 LLM 直接生成完整的文件内容而是让它生成 diff 或者指定替换范围。否则大文件场景下 token 消耗会失控。async def execute_node(state: AgentState): step state[plan][state[current_step]] prompt f当前步骤{step} 可用工具{tool_descriptions} 相关代码上下文{state[code_context]} 请返回要调用的工具名和参数JSON 格式。 response await llm.ainvoke(prompt) tool_call parse_json(response) result await dispatch_tool(tool_call) return {execution_result: result, messages: [response]}3.3 验证节点与条件边让 Agent 自己判断成败验证节点负责检查上一步执行结果。如果是run_command执行测试就看返回码如果是文件写入就检查文件是否存在且语法正确。验证不通过时LangGraph 的条件边会把流程导回执行节点同时error_count加一。def should_continue(state: AgentState) - str: if state[error_count] 3: return fail if state[execution_result] 表示成功: if state[current_step] len(state[plan]) - 1: return finish return next_step return retry这张条件边的设计直接决定了 Agent 的鲁棒性。我踩过的坑是早期版本没有error_count上限结果一个依赖缺失的问题让 Agent 反复重试了二十多次token 账单直接爆炸。4. FastAPI 服务层与 LangGraph 的对接细节4.1 接口设计与流式返回羲和对外暴露两个核心接口POST /agent/run接收任务并启动执行GET /agent/stream/{session_id}用 SSE 推送执行进度。为什么用 SSE 而不是 WebSocket因为编码任务的执行过程是单向推送——服务端不断把节点执行结果推给客户端客户端不需要中途发消息。SSE 实现更简单FastAPI 原生支持StreamingResponse。from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() app.post(/agent/run) async def run_agent(req: TaskRequest): session_id create_session(req.task) asyncio.create_task(agent_graph.ainvoke(...)) return {session_id: session_id} app.get(/agent/stream/{session_id}) async def stream(session_id: str): async def event_generator(): async for event in session_queue[session_id]: yield fdata: {json.dumps(event)}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)4.2 会话状态持久化LangGraph 自带 checkpointer 机制可以把每一步的状态存到 SQLite 或 Postgres。我在services/session.py里封装了一层用AsyncSqliteSaver做本地持久化。这样即使用户关掉页面再回来通过 session_id 依然能查到之前的执行记录。提示checkpointer 的thread_id要和你的 session_id 对应起来否则恢复状态时会找不到对应的执行线程。4.3 配置管理与环境隔离所有敏感配置走.env文件通过 Pydantic 的BaseSettings加载。LLM 的 API key、模型名称、超时时间都放在core/config.py里统一管理。我习惯给每个环境准备独立的.env文件本地开发用gpt-4o-mini降低成本生产环境再切到更强的模型。5. 常见问题与排查技巧实录5.1 节点执行卡死或超时最常见的原因是run_command执行的命令进入了交互模式比如pip install等待确认。解决方法是在命令前加-y参数或者在工具实现里设置timeout并用subprocess的communicate(timeout...)强制中断。5.2 LLM 返回格式不符合预期LangGraph 节点之间的数据传递依赖结构化输出但 LLM 偶尔会“自由发挥”。我的做法是在每个节点的 prompt 末尾加一句“只输出 JSON不要任何解释”同时在解析层做兜底——解析失败时把原始返回当作纯文本处理并记录一条 warning 日志。5.3 状态字段被意外覆盖LangGraph 的默认行为是节点返回的字典会合并到现有状态中但如果你返回的字段类型和原类型不一致会直接报错。比如messages字段必须用add_messagesreducer手动返回列表会覆盖历史消息。这个坑我在早期版本踩过导致多轮对话上下文丢失。5.4 排查速查表现象可能原因解决方向Agent 无限循环缺少 error_count 熔断加最大重试次数文件修改后不生效写入路径错误打印绝对路径核对SSE 连接断开代理缓冲了流加X-Accel-Buffering: no头状态恢复失败thread_id 不匹配检查 session 映射关系命令执行无输出子进程缓冲加-u或 flush6. 从问答到执行的关键设计心得把 LangGraph 的planning 模式真正跑通核心不在于图有多复杂而在于每个节点的职责边界要清晰。规划节点只负责拆步骤执行节点只负责调工具验证节点只负责判断成败。我见过有人把规划和执行揉在一个节点里结果 prompt 越写越长LLM 的输出越来越不稳定。另一个心得是工具的描述比工具的代码更重要。LLM 决定调用哪个工具完全依赖你给的工具描述。read_file的描述写成“读取文件”和写成“读取指定路径的文件内容支持指定起止行号用于查看代码上下文”后者的调用准确率明显更高。这套架构目前支撑了羲和的日常使用从简单的“解释这段代码”到“帮我重构这个模块并跑通测试”都能覆盖。后续如果要扩展我建议优先加一个“回滚节点”——当 Agent 连续失败时自动恢复到任务开始前的文件状态避免把项目改得面目全非。这个功能在实际使用中比想象中更必要。

FastAPI+LangGraph实战：从零构建可执行任务的编码助手

相关推荐

2026年代理服务测评与高风控场景实战策略

Java网络编程：Socket套接字与TCP/UDP实战

STM32F303基础工程文件入门：从零到点灯与避坑指南

更多资讯

VC2015编译libssh-0.10.3静态库：Windows下SFTP集成确定性方案

土木工程师转型网络安全的路径与实战分析

Java数据结构选型指南：从原理到性能优化实战

动态规划与贪心算法在多边形周长问题中的应用

VS Code SFTP插件实现跨平台文件同步实战

Spring Boot应用Docker容器化与Nginx部署实践

今日更新

AI辅助技术栈评估：提升研发决策效率60%

从生成Token到直接决策：Jev如何重塑AI Agent工程实践

32G统一内存跑大模型：内存账、TPS实测与端云分流

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？