智能体Agent核心模块Graph图经典代码分析-学经典掌握智能体开发

图Graph通俗说法是串起来的线，有开始有结束，让数据在这条线上流动处理，最终有结果。把代码翻看一遍后，我用编程新手角度理解是里面合适地使用了类似函数指针技术，调度了一系列函数。当然这只是通俗易懂的比喻，让大家建立记忆印象入口，并不是说图Graph就是简单的如此，图Graph是智能体Agent核心模块，技术有质的突破，非常复杂，理论术语理解有困难，请联系锐英源软件。

智能体Agent的项目包含模块比较多，本文是系列文章，系列里介绍各个模块，希望看到的IT类朋友关注下锐英源软件，收藏本文，关注系列。

代码摘自liquid-quality-agent，liquid-quality-agent把工业视觉检测和Agent结合，项目经典，语言是Python语言。

理论

Agent中的Graph（图）是其执行核心，它将传统Agent的线性或简单循环逻辑，升级为一个高度可控、有状态、可灵活编排的智能工作流。
它的核心思想是，将Agent的思考与行动过程，拆解为一张由节点（Nodes）、边（Edges） 和状态（State） 构成的有向状态图（StateGraph）。
🎯 Graph要实现什么？
Graph的核心目标是让Agent的决策过程变得可控、可观测且可追溯。与传统的线性链（Chain）相比，Graph通过状态图模型，能实现更复杂的逻辑：

复杂流程编排：支持条件分支、循环、并行执行等，能处理需要多轮“思考-行动”的复杂任务。
状态持久化与记忆：通过全局State在节点间传递信息，实现短期工作记忆和长期记忆。
增强可观测性：将执行流程可视化，每个节点的输入输出都清晰可查，便于调试和优化。

🏗️ 一般架构：三大核心组件
一个LangGraph Agent的架构由以下三个核心组件构成：

状态 (State)：一个在节点间流动的共享数据结构，是Graph的“工作内存”。它通常是一个TypedDict或Pydantic模型，定义了Graph的输入输出格式，所有节点的读写都基于此。
节点 (Nodes)：Graph中的执行单元，是具体的Python函数。每个节点接收当前的State，执行特定任务（如调用LLM、执行工具、处理数据），然后返回更新后的State

边 (Edges)：定义了节点间的连接与执行流向。

固定边：从一个节点无条件指向下一个节点。
条件边：根据当前State动态决定下一个要执行的节点，是实现if-else分支和循环的关键。

一个Graph在运行前需要先编译（Compile），这个过程会检查图结构合法性，并配置检查点（Checkpointer）等运行时参数。
🔄 输入与输出
Graph的输入输出由其定义的State结构决定。

输入 (Input)：启动Graph时，传入一个符合State定义的字典。一个典型输入至少包含用户的任务或问题（通常在"messages"字段中）。
输出 (Output)：Graph执行完成后，返回最终状态的字典。它通常包含：

最终答案：Agent生成的最终回复。
完整消息历史：记录了完整的“思考”与“行动”轨迹。
中间步骤与证据：包含调用工具的结果、检索到的文档等，用于支撑最终结论。

💡 一个典型的工作流示例
一个基于LangGraph的Agent工作流可能如下：

入口 (START)：接收用户输入，初始化State。
理解节点 (LLM)：LLM分析意图，决定是直接回答、检索知识还是调用工具。
条件边：根据LLM的决策，路由到不同节点。
执行节点 (Action)：若需调用工具，则执行相应工具（如搜索、计算）并更新State。
结果处理节点 (Observation)：处理工具返回结果，更新State。
循环或结束：通过条件边判断任务是否完成，未完成则返回步骤2继续，完成则进入结束节点。
结束 (END)：整理State中的信息，生成最终答案输出。

总的来说，Graph提供了一种结构化的方式，将复杂的决策过程拆解为清晰、可控的步骤，是实现可靠、可解释Agent的关键。

框架接口

api.py里应该是接口代码，但是Graph怎么和其它模块结合的？应该是在http请求调用上下文里。

get_analysis_graph是图的生成函数，搜索get_analysis_graph，出现在下面位置：
@app.post("/agent/analyze", tags=["Agent"])
async def analyze(request: AgentRequest) -> dict:
try:
return await run_in_threadpool(
lambda: get_analysis_graph().invoke(
request.record_id,
request.request,
)
)
except Exception as exc:
raise HTTPException(
status_code=500,
detail=f"{type(exc).__name__}: {exc}",
) from exc

上面是函数代码，下面介绍一个重点细节：
lambda: get_analysis_graph().invoke(
加 lambda 的核心目的只有一个：将“立即执行”转变为“延迟执行”（惰性求值）。
如果没有 lambda，get_analysis_graph().invoke(...) 会在定义时（传参时）立即被执行，并把执行结果（返回值）传给外层函数。加上 lambda 后，这段代码变成了一个可调用的函数对象，只有在外层函数内部主动调用这个 lambda 时，.invoke() 才会真正运行。
具体来说，这行代码通常出现在以下典型场景中：
1. 异步任务/线程池提交（最常见）
在使用 ThreadPoolExecutor 或 Celery 时，提交任务不能立即执行，而是要交给调度器。
python
# 错误：立即执行，submit 拿到的是执行后的结果（而非任务）
executor.submit(get_analysis_graph().invoke(request.record_id, request.request))

# 正确：延迟执行，submit 拿到的是任务函数，调度器决定何时调用 lambda
executor.submit(lambda: get_analysis_graph().invoke(request.record_id, request.request))
2. 重试机制或条件触发
如果你正在使用 tenacity 重试库，或是某些工作流引擎（如 LangGraph 的 Command 或节点回调），框架需要传入一个“动作”而不是“结果”，以便在失败时重新执行。加 lambda 使得每次调用都会重新获取 get_analysis_graph() 实例，确保每次尝试都是全新的执行上下文。
3. 闭包捕获变量（按值引用）
虽然代码里直接用了 request，但加上 lambda 形成了一个闭包。这在循环或异步回调中特别重要——它保证执行时能正确拿到当前的 request 对象，而不会因为外部变量变化导致预期外的结果（虽然在当前单行调用中这不是主因，但这是 lambda 的通用特性）。
4. 避免无谓的资源开销
如果 get_analysis_graph() 初始化开销很大（例如加载模型权重），直接执行会立即占用资源。用 lambda 包装后，只有在真正需要分析时（例如用户点击按钮或满足特定业务条件）才会创建 Graph 对象并调用 invoke，实现了按需加载。

简单对比：

不加 lambda：result = get_analysis_graph().invoke(...) → 变量 result 存放的是分析结果（数据）。
加 lambda：task = lambda: get_analysis_graph().invoke(...) → 变量 task 存放的是分析动作（函数）。只有当你写 task() 时，分析才会发生。

analysis_graph: QualityAnalysisGraph | None = None
record.record_id,
request.request,
_authoritative_decision_basis分析关键信息

Graph围绕AgentState，使用里面的属性，在图里链条上不同函数里填充属性，并修改属性，这样Graph就有了结果。

关键代码

D:\liquid-quality-agent\py311_env\Lib\site-packages\langgraph\graph，基类库提供的，在Anaconda环境里，是前辈们实现过的。

图的初始化：构造

def _build(self):
builder = StateGraph(AgentState)
builder.add_node("load_record", self._load_record)
builder.add_node("retrieve_knowledge", self._retrieve_knowledge)
builder.add_node("generate_analysis", self._generate_analysis)
builder.add_node("validate_output", self._validate_output)
builder.add_edge(START, "load_record")
builder.add_conditional_edges(
"load_record",
self._record_route,
{"found": "retrieve_knowledge", "missing": END},
)
builder.add_edge("retrieve_knowledge", "generate_analysis")
builder.add_edge("generate_analysis", "validate_output")
builder.add_edge("validate_output", END)
return builder.compile()

图的调用：

def invoke(self, record_id: str, request: str) -> dict:
state = self.graph.invoke(
{"record_id": record_id, "request": request}
)
return state["result"]
对图概念理解下，实现不困难。调用是调用build里构造的链条上的各类函数，从load_record开始，到validate_output结束。

对大模型调用

1. with_structured_output 方法

structured_model = self.model.with_structured_output(
AgentAnalysis,
method="function_calling",
include_raw=True,
)

self.model：这是一个 LangChain 的聊天模型实例（如 ChatOpenAI），具备调用能力。
with_structured_output：LangChain 提供的方法，用于将普通聊天模型“包装”成一个能够输出固定结构数据的模型。它要求模型支持函数调用（Function Calling）或 JSON 模式。
AgentAnalysis：这是一个 Pydantic 模型（或 dataclass），定义了期望输出的字段、类型和验证规则。模型会按照这个 schema 生成符合要求的数据结构。
method="function_calling"：指定实现结构化输出的方式为“函数调用”。对于 OpenAI 等模型，这会在请求中声明一个 function，模型会返回一个包含函数参数的 JSON，从而保证输出格式严格匹配。
include_raw=True：使返回结果不仅包含解析后的 AgentAnalysis 对象，还会包含原始模型响应（如 raw 字段包含完整的 API 返回内容），便于调试或后续处理。

经过包装后，structured_model 的 invoke 方法将自动处理输出解析。

2. 调用模型：structured_model.invoke(...)

response = structured_model.invoke(
[
SystemMessage(content=SYSTEM_PROMPT),
HumanMessage(
content=json.dumps(
{
"request": state["request"],
"record": self._compact_record(record),
"knowledge": knowledge,
},
ensure_ascii=False,
)
),
]
)

invoke：执行模型调用，传入的消息列表包含两条：

SystemMessage：系统提示词，内容为 SYSTEM_PROMPT（通常包含角色定义、输出格式要求等）。
HumanMessage：用户消息，内容是一个 JSON 字符串，由三个字段组成：

"request"：来自状态 state["request"]，可能表示用户的原始请求或问题。
"record"：经过 self._compact_record(record) 压缩后的记录数据（可能用于减少 token 消耗或提取关键信息）。
"knowledge"：外部知识库或上下文信息，辅助模型做出分析。

ensure_ascii=False 保证 JSON 中的非 ASCII 字符（如中文）不被转义，提高可读性。

模型根据系统指令和提供的 JSON 上下文，生成符合 AgentAnalysis 结构的输出。由于启用了函数调用，模型会“调用”预定义的函数，其参数即为 AgentAnalysis 所需字段。

3. 返回值 response
由于设置了 include_raw=True，response 通常是一个字典或对象，包含：

parsed：解析后的 AgentAnalysis 实例（可直接访问字段）。
raw：模型原始的响应对象（如 ChatGeneration 或 message）。
parsing_error（可选）：如果解析失败，包含错误信息。

后续代码可以从 response.parsed 中提取结构化分析结论，用于决策或进一步处理。

整体作用
这段代码是智能体（Agent）工作流中的一个核心环节：将用户请求、压缩后的记录和外部知识融合，交由 LLM 进行结构化分析，并以可编程的方式获得分析结果。这种方式避免了从自由文本中人工解析的麻烦，保证了输出的可靠性和一致性，非常适合用于质量检测、分类、决策等自动化任务。

潜在上下文补充

_compact_record(record) 可能是对原始数据（如传感器读数、日志）进行摘要或裁剪，保证输入长度适配模型上下文。
state 和 knowledge 来源于上游节点，由状态管理（如 LangGraph 的 State）传递。
该图（Graph）可能属于一个多节点流程，此节点负责调用 LLM 分析液体质量，并输出结构化的结论（如是否合格、异常原因等）。

友情链接