LangGraph + Ertas

使用 LangGraph 构建生产级智能体——这是一个基于图的编排框架，提供审计跟踪、检查点和有状态工作流，与 Ertas 训练的本地模型完全兼容。

Overview

LangGraph 是 LangChain 智能体的生产级继任者，专为在生产中运行智能体的运营需求而设计：显式状态机、持久化检查点、审计跟踪和人机协同中断点。LangChain 较早期的智能体抽象使用隐式的 ReAct 风格循环，而 LangGraph 要求你将智能体定义为状态和转换的有向图——这是一个前期成本，但在生产条件下，可观测性、可恢复性和可靠性方面会带来显著改进，从而获得回报。到 2026 年初，LangGraph 在 GitHub 星标上超越了 CrewAI，现在是 Python 生态系统中占主导地位的生产级智能体框架。

基于图的设计能够实现扁平智能体循环难以提供的能力：具有结构化聚合的并行分支执行、基于中间状态的条件路由、允许智能体跨越数小时或数天暂停和恢复的持久化检查点，以及让你能够从任何状态重放智能体运行的回滚点。对于需要处理长时运行工作流、集成人工审批步骤或从基础设施故障中恢复的企业部署，LangGraph 越来越成为优于更简单替代方案的默认选择。

How Ertas Integrates

Ertas 训练的模型通过与 LangChain 相同的提供商接口与 LangGraph 协同工作——这两个框架共享其底层 LLM 抽象。在 Ertas Studio 中微调模型并部署到兼容 OpenAI 的端点（Ollama、vLLM 或 Ertas Cloud）后，你可以配置一个指向你的端点的 LangChain ChatOpenAI 或 ChatOllama 实例，然后将该 LLM 传递给你的 LangGraph 节点。每个调用 LLM 的节点都使用你的 Ertas 训练模型，而 LangGraph 的编排功能（检查点、分支、人机协同）无论 LLM 接口背后是哪种模型，都能透明地工作。

对于生产部署，LangGraph + Ertas 组合同时解决两个关注点。LangGraph 处理运营关注：工作流持久性、审计日志和可靠性。Ertas 处理质量关注：领域特定的模型专门化，提供比通用模型显著更好的任务性能。结果是既经过良好工程化（持久、可观测、可恢复）又经过良好训练（领域专门化、准确、快速）的智能体。对于在受监管行业（医疗保健、金融、法律）构建智能体的团队来说，这种组合尤其有吸引力，因为 LangGraph 的审计能力和 Ertas 的本地部署相结合，可以满足大多数合规性要求而无需妥协。

Getting Started

1
在 Ertas Studio 中微调领域模型
训练一个捕获你的领域专业知识的模型。具体来说，对于 LangGraph 工作流，包含显式推理步骤和工具使用轨迹的训练数据会产生最佳结果。
2
部署到兼容 OpenAI 的端点
导出为 GGUF 格式并通过 Ollama、vLLM 或 Ertas Cloud 提供服务。LangGraph 通过 LangChain 提供商抽象调用任何标准的聊天补全端点。
3
安装 LangGraph 并配置你的 LLM
安装 langgraph 和 LangChain 提供商包（langchain-openai 或 langchain-ollama）。创建一个指向你的 Ertas 推理端点的 LLM 实例。
4
将你的智能体定义为图
将你的工作流建模为有向图：每个步骤的节点、转换的边以及用于路由的条件逻辑。在需要的地方添加用于持久化状态和人机协同中断的检查点。
5
通过 LangGraph Cloud 或自托管部署
将你的图部署到 LangGraph Cloud 进行托管编排，或使用 LangGraph Server 自托管。两种选项都适用于自托管的 Ertas 推理。

python

from typing import TypedDict, Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# Point LangGraph at your Ertas-trained model
llm = ChatOpenAI(
    base_url="http://localhost:11434/v1",
    model="ertas-claims-adjuster-14b",
    api_key="not-needed",
    temperature=0.1,
)

class ClaimState(TypedDict):
    claim_id: str
    claim_data: dict
    classification: str
    requires_review: bool
    decision: str

def classify(state: ClaimState):
    response = llm.invoke(f"Classify this claim: {state['claim_data']}")
    return {"classification": response.content}

def route(state: ClaimState) -> Literal["auto_approve", "human_review"]:
    return "human_review" if state["requires_review"] else "auto_approve"

# Build the graph
graph = StateGraph(ClaimState)
graph.add_node("classify", classify)
graph.add_node("auto_approve", lambda s: {"decision": "approved"})
graph.add_node("human_review", lambda s: {"decision": "pending_review"})

graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route, {
    "auto_approve": "auto_approve",
    "human_review": "human_review",
})
graph.add_edge("auto_approve", END)
graph.add_edge("human_review", END)

# Compile with persistent checkpointing
app = graph.compile(checkpointer=MemorySaver())

# Run with full audit trail
result = app.invoke(
    {"claim_id": "C-12345", "claim_data": {...}},
    config={"configurable": {"thread_id": "claim-12345"}},
)

在 LangGraph 中构建一个由 Ertas 训练领域模型支持的有状态保险理赔处理智能体，带有完整的审计跟踪和持久化检查点。