1. 为什么我要从零造一个编码助手
市面上不缺代码问答工具，缺的是能把"说"变成"做"的东西。我最初做羲和（XiheAgent）的动机很朴素：日常开发里大量重复性任务——批量改接口签名、给新模块补单元测试、根据报错日志定位并修复问题——这些事用纯对话式助手做，我得手动复制粘贴、来回确认，效率提升有限。真正想要的是一个能读代码库、能规划步骤、能调用工具、能自己验证结果的执行体。
羲和这个名字取自上古掌管天文历法的神，寓意"观测与调度"。它的定位很明确：一个基于 FastAPI 提供接口、用 LangGraph 编排任务流的 AI 编码助手，核心能力覆盖代码问答、任务规划、工具调用、结果校验四个层次。适合谁参考？如果你已经会写 Python、了解基本的 LLM 调用，但卡在"怎么把一次问答升级成多步任务执行"这个坎上，那这篇内容基本就是我想跟你聊的东西。
我踩过的第一个坑就是：一开始用最朴素的 while 循环 + if-else 来串流程，结果状态管理一团乱，重试逻辑和分支判断缠在一起，改一处崩三处。后来换成 LangGraph 才把"节点 + 边 + 状态"这套模型理顺。所以下面我会重点讲清楚：为什么选 LangGraph 而不是裸写循环，FastAPI 的目录怎么分层才不烂尾，以及 planning 模式在编码任务里到底怎么落地。
2. LangGraph 到底解决了裸写循环的哪些痛点
2.1 从"一问一答"到"多步执行"的本质变化
纯问答的链路是线性的：用户输入 → 拼 prompt → 调模型 → 返回。但任务执行不是线性的。举个真实例子，用户说"帮我把 user 模块里所有同步数据库调用改成异步"。这个任务至少包含：扫描文件、识别同步调用、判断哪些能安全改、逐个改写、跑测试验证、失败回滚。中间任何一步都可能需要回头重做，比如测试挂了要回到改写步骤。
裸写循环的问题在于，这些"回头"逻辑会变成嵌套的 if-else 和 goto 式的跳转，状态散落在各种局部变量里。LangGraph 的核心价值就是把流程显式建模成图：节点是执行单元，边是流转规则，状态是一个贯穿始终的共享数据结构。这样"回头重做"就变成了一条从验证节点指回改写节点的边，逻辑一目了然。
2.2 LangGraph 和 LangChain 的区别，别再搞混
这是被问得最多的问题。我的理解是：LangChain 是工具箱，LangGraph 是流水线控制器。LangChain 提供的是 LLM 调用、Prompt 模板、输出解析、检索器这些"零件"；LangGraph 提供的是把这些零件按有向图组织起来、并管理状态流转的"骨架"。
在羲和里，两者是配合使用的：用 LangChain 的组件做单次模型调用和工具封装，用 LangGraph 的 StateGraph 做整体编排。如果你只做单轮问答，LangChain 的 Chain 就够了；一旦涉及条件分支、循环重试、多角色协作，就该上 LangGraph。面试里常问的"什么时候用 LangGraph"，我的标准回答是：当你的流程需要根据中间结果动态决定下一步走向时。
2.3 状态设计：整个图的中枢神经
LangGraph 里最需要想清楚的就是 State。它本质上是一个 TypedDict，定义了所有节点共享的数据字段。羲和的 State 大致长这样：
from typing import TypedDict, Annotated, Sequence from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[Sequence, add_messages] # 对话历史，自动追加 task_plan: list[str] # 规划出的步骤列表 current_step: int # 当前执行到第几步 code_context: dict # 读取到的代码片段 tool_results: list[dict] # 工具调用结果 retry_count: int # 当前步骤重试次数 final_answer: str # 最终输出这里有个关键细节：messages字段用了Annotated[..., add_messages]，这是 LangGraph 的 reducer 机制。普通字段是"覆盖"语义，新值直接替换旧值；而加了 reducer 的字段是"合并"语义，新消息会追加到列表里而不是覆盖。我一开始没注意这点，导致对话历史每次都被清空，排查了半天才发现是 reducer 没配。
提示：状态字段能少则少，只放真正需要跨节点共享的数据。把临时变量塞进 State 会让图变得难以调试。
3. FastAPI 项目目录结构：分层没做好，后期全是债
3.1 我推荐的目录骨架
FastAPI 项目最容易犯的错就是所有路由堆在一个 main.py 里。羲和跑起来之后接口有十几类，如果一开始不分层，后面加个鉴权、加个日志中间件都会牵一发动全身。我最终稳定下来的结构是这样的：
xihe_agent/ ├── app/ │ ├── main.py # 应用入口，注册路由和中间件 │ ├── api/ │ │ ├── deps.py # 依赖注入（数据库、鉴权、Agent 实例） │ │ └── v1/ │ │ ├── chat.py # 问答相关接口 │ │ ├── task.py # 任务执行接口 │ │ └── health.py # 健康检查 │ ├── core/ │ │ ├── config.py # 配置管理（pydantic-settings） │ │ └── logging.py # 日志配置 │ ├── graph/ │ │ ├── state.py # AgentState 定义 │ │ ├── nodes.py # 各个节点函数 │ │ ├── builder.py # 图构建与编译 │ │ └── tools.py # 工具定义 │ ├── schemas/ │ │ └── request.py # 请求/响应模型 │ └── services/ │ └── agent_service.py # 业务逻辑，隔离路由和图 ├── tests/ └── pyproject.toml分层的核心原则是路由只负责收参数和返结果，业务逻辑放 service，图编排放 graph。这样测试的时候可以单独测节点函数，不用起整个 HTTP 服务。
3.2 依赖注入怎么用才不别扭
FastAPI 的Depends是它最优雅的设计之一。羲和里我用它来管理 Agent 实例的生命周期——图编译一次比较耗时，不能每个请求都重新编译。做法是在应用启动时编译好图，存到 app.state，然后通过依赖注入取出来：
# app/api/deps.py from fastapi import Request def get_agent(request: Request): return request.app.state.agent # app/api/v1/task.py @router.post("/execute") async def execute_task(payload: TaskRequest, agent=Depends(get_agent)): result = await agent.ainvoke({"messages": [payload.query]}) return {"answer": result["final_answer"]}这里有个坑：LangGraph 编译出来的图对象是有状态的，如果多个请求并发调用同一个图实例，状态可能串。我的处理方式是每个请求用独立的初始 State，图本身只读不写，这样并发是安全的。如果你在节点里改了全局变量，那就要小心了。
3.3 配置管理别硬编码
API Key、模型名、超时时间这些绝对不能写死在代码里。我用 pydantic-settings 做配置，支持从环境变量和 .env 文件读取：
from pydantic_settings import BaseSettings class Settings(BaseSettings): model_name: str = "gpt-4o" api_key: str max_retry: int = 3 request_timeout: int = 60 class Config: env_file = ".env" settings = Settings()好处是类型安全，启动时就能发现配置缺失，而不是跑到一半才报错。
4. Planning 模式在编码任务里的落地方式
4.1 为什么编码任务特别需要先规划
编码任务有个特点：步骤之间有强依赖，且错误会累积。如果让模型边想边做，第一步理解错了，后面全错。Planning 模式的核心思想是先把任务拆成有序的步骤列表，再逐步执行，每步执行完可以校验。
在 LangGraph 里，这对应一个典型的"规划节点 → 执行节点 → 校验节点"的循环结构。规划节点负责把用户意图拆解成步骤，执行节点负责做当前步骤，校验节点判断是否成功、是否需要重试或重新规划。
4.2 规划节点的 prompt 怎么写
规划节点的输出必须是结构化的步骤列表，不能是自由文本。我用的是让模型输出 JSON 数组，然后用 Pydantic 校验：
PLANNER_PROMPT = """你是一个编码任务规划器。把用户的任务拆解成可执行的步骤。 每个步骤必须是一个具体的动作，不要有模糊描述。 输出格式为 JSON 数组，例如： ["读取 user.py 文件", "定位所有同步 db 调用", "逐个改写为异步"] 只输出 JSON，不要有其他内容。"""实测下来，让模型"只输出 JSON"比"请以 JSON 格式输出"效果好很多，后者经常会在 JSON 前后加解释性文字。解析的时候我还会加一层容错，用正则提取第一个[...]块，防止模型偶尔不听话。
4.3 执行与校验的循环怎么控制
执行节点从task_plan里取current_step对应的步骤，调用相应工具，把结果写进tool_results。校验节点判断这一步是否成功。这里的关键是重试上限和重新规划的条件要分开：
- 单步失败：
retry_count加一，没超过max_retry就重试当前步骤。 - 单步反复失败：说明规划本身有问题，触发重新规划，把失败信息喂回规划节点。
- 全部步骤完成：进入汇总节点，生成最终答案。
用 LangGraph 的条件边来表达就是：
def should_continue(state: AgentState) -> str: if state["current_step"] >= len(state["task_plan"]): return "summarize" if state["retry_count"] >= settings.max_retry: return "replan" return "execute" graph.add_conditional_edges("validate", should_continue, { "summarize": "summarize", "replan": "planner", "execute": "executor", })这种写法比裸写循环清晰太多，每个分支的走向一眼可见。
5. 工具调用：让助手真正"动手"的关键
5.1 工具的定义与注册
编码助手的工具无非几类：读文件、写文件、执行命令、搜索代码。在 LangGraph 里，工具可以用 LangChain 的@tool装饰器定义，然后绑定到模型上：
from langchain_core.tools import tool @tool def read_file(path: str) -> str: """读取指定路径的文件内容""" with open(path, "r", encoding="utf-8") as f: return f.read() @tool def search_code(keyword: str, root: str = ".") -> list[str]: """在代码库中搜索包含关键词的文件""" import subprocess result = subprocess.run( ["grep", "-rl", keyword, root], capture_output=True, text=True ) return result.stdout.strip().split("\n")工具函数的 docstring 非常重要，模型就是靠它来判断什么时候该调用哪个工具。我踩过的坑是 docstring 写得太笼统，导致模型该调read_file的时候去调了search_code。后来把每个工具的适用场景写清楚，准确率明显提升。
5.2 工具执行的安全边界
让 AI 直接执行命令是有风险的。我的做法是加一层白名单和沙箱：只允许执行测试、格式化、静态检查这类命令，禁止任何删除、网络请求、系统修改类操作。写文件也限制在项目目录内，路径要做规范化校验，防止../越界。
ALLOWED_COMMANDS = {"pytest", "ruff", "mypy", "black"} def safe_execute(cmd: list[str]): if cmd[0] not in ALLOWED_COMMANDS: raise PermissionError(f"命令 {cmd[0]} 不在白名单内") return subprocess.run(cmd, capture_output=True, text=True, timeout=30)注意：白名单要按前缀匹配而不是精确匹配，否则
pytest -k test_foo这种带参数的调用会被误拦。
5.3 工具结果的压缩与回传
工具返回的内容可能很长，比如读一个大文件。如果原样塞回上下文，token 消耗会爆炸。我的处理是：读文件时只返回相关片段（用关键词定位前后若干行），命令输出只保留末尾若干行和错误信息。这样既保留了关键信息，又控制了上下文长度。
6. 实测中那些文档不会告诉你的坑
6.1 状态字段的 reducer 陷阱
前面提过add_messages的 reducer，这里再展开说。LangGraph 里如果两个节点并行执行、都要写同一个字段，没有 reducer 就会报错或覆盖。我一开始把tool_results定义成普通 list，结果并行节点写的时候互相覆盖。改成带 reducer 的写法：
from operator import add tool_results: Annotated[list, add]这样多个节点的结果会自动拼接。记住一个原则：会被多个节点写入的字段，一定要配 reducer。
6.2 无限循环的防范
条件边如果写错，图可能陷入死循环。LangGraph 编译时可以设置recursion_limit：
app = graph.compile() result = app.invoke(state, config={"recursion_limit": 25})超过限制会抛异常。我建议在开发阶段把这个值设小一点（比如 15），这样一旦逻辑有问题能快速暴露，而不是等到跑了几百步才发现。
6.3 流式输出与图执行的配合
用户等一个多步任务执行完可能要几十秒，没有流式反馈体验很差。FastAPI 里可以用StreamingResponse配合图的astream方法，每个节点执行完就推一次进度：
async def stream_task(query: str): async for event in agent.astream({"messages": [query]}): for node_name, node_output in event.items(): yield f"data: {node_name} 完成\n\n"前端用 EventSource 接收，就能实时看到"正在规划""正在执行第 2 步"这样的进度。这个体验提升非常明显，用户不会觉得程序卡死了。
6.4 模型输出不稳定的兜底
再好的 prompt 也不能保证模型每次都输出合法 JSON。我的兜底策略是三层：第一层用 Pydantic 校验，第二层用正则提取，第三层失败就降级成让模型重新输出一次。三层都失败才报错。实测下来，加了兜底之后任务成功率从大概七成提到了九成以上。
7. 从问答到执行，我个人的几点体会
做羲和这段时间，最大的感受是：AI 编码助手的难点不在模型，而在工程。模型能力是现成的，但怎么把它的输出可靠地接入真实代码库、怎么管理多步状态、怎么在出错时优雅降级，这些才是决定产品能不能用的关键。
LangGraph 给我的最大启发是"把控制流显式化"。以前写 Agent 总想着用一个大 prompt 让模型自己决定一切，结果不可控也不可调试。现在把流程拆成节点和边，每个环节的输入输出都清清楚楚，出问题能定位到具体节点，改起来也有针对性。
如果你准备动手做类似的东西，我的建议是：先把最简单的"规划 → 执行 → 校验"三节点跑通，别一上来就搞复杂的多 Agent 协作。等这条主线稳定了，再往上加工具、加流式、加重试策略。工程这东西，能跑通的最小闭环永远比设计完美的架构更重要。
最后分享一个调试小技巧：LangGraph 支持把执行过程可视化，开发阶段我会把每次运行的图结构导出看一眼，确认边的走向符合预期。很多逻辑 bug 光看代码看不出来，画出来一眼就发现了。
