工具做事,图负责调度
第 3 节:Tool 与 LangGraph 工作流
把数据能力封装为 Tool,再用 StateGraph 控制先后与分支。
1. 工具返回什么
本地工具用 @tool 声明,输入参数明确。行情与搜索数据都标注模拟,不会连接网络。真实项目可替换函数内部实现,但要保留来源、时间和异常处理。
@tool
def mock_search(company: str) -> list[dict]:
"""查询本地示例资料,不会访问互联网。"""
return [{"label": "模拟资料", "title": "虚构条目", "url": "https://example.invalid/mock-source"}]2. State 是节点共享的当前任务
问题解析结果写入 query,工具结果写入 quote 或 sources,回答节点读取这些字段。节点只返回本次更新内容;图负责合并状态。
class AgentState(TypedDict, total=False):
question: str
query: dict
quote: dict | None
sources: list[dict]
answer: str
errors: list[str]3. 条件路由选择下一步
没有识别公司时直接说明;询问行情就进入行情节点;需要最新资料则进入搜索。行情与搜索都需要时,先查行情再查资料。节点运行完进入回答节点。
graph.add_edge(START, "parse")
graph.add_edge("parse", "cache")
graph.add_conditional_edges("cache", route, {
"quote": "quote", "search": "search", "answer": "answer",
})
graph.add_conditional_edges("quote", after_quote, {
"search": "search", "answer": "answer",
})
graph.add_edge("search", "answer")
graph.add_edge("answer", END)
app = graph.compile()4. LangChain 和 LangGraph 分工
LangChain 提供模型、Prompt、结构化输出和工具接口;LangGraph 用 State、节点、边和条件路由表达这个项目的工作流。完整实现都在第 1 节代码中。