LangChain 和 LangGraph

查看 Markdown查看 Markdown

LangChain create_agent 在 LangGraph 上构建会调用 tool 的 Agent。Dex adapter 把 create_agent Agent 或自定义的 LangGraph graph 放进 durable session 中运行。它支持 Python。

何时使用

Agent 已经用 LangChain create_agent 或 LangGraph graph 编写时，使用它。保留 graph、它的 middleware 和 tool，由 Dex 管理 session。

Session 如何运行

DurableToolCallMiddleware 用 LangGraph interrupt 拦下每个 tool call，因此 RunModel 内不会运行任何 tool。RunModel Step 把 graph 运行到它的 tool call 为止，通常只调用一次模型，并保存 graph checkpoint。

每个 tool call 在自己的 Step 中并行运行：read 或 idempotent write 用 CallTool，其他写操作用 CallSingleAttemptTool。需要审批的调用先在 AwaitToolApproval 中等待。Tool Step 校验参数并调用 tool 一次。下一个 RunModel 恢复 checkpoint，并用记录下来的结果继续运行 graph。

自定义 graph 用 durable_tool_node 代替 ToolNode 构建 tool node。它的 state 必须以 LangChain message 保存对话。

查询订单的 turn 依次运行 RunModel、CallTool、ResumeTurn 和 RunModel。

安装

把 adapter 与它的 framework 一起安装。每个 adapter 都会带上 Dex Agents core。

$

pip install dex-agents-langgraph langchain-openai

例子: agents/python/langgraph/README.md

Example

这个 example 就是每个 framework 页面共用的订单客服 Agent。

Flow type 是 LangChainOrderSupport（Python）。完整 example 位于 examples/python/agent-harnesses。

Agent 与 tool

Tool 是 LangChain tool。一个工厂函数用 DurableToolCallMiddleware 构建 Agent。只有 Worker 会调用这个工厂函数，因此 Client 进程不需要 provider key。Tool policy 指定每个 tool。

langchain_order_support.pyPython

123456789101112131415161718192021222324252627282930313233343536

def build_langchain_order_support_flow(
    order_desk: OrderDesk,
    model: str | BaseChatModel,
    slack_connection: slack.Connection | None,
) -> AgentSessionFlow:
    """Builds the session Flow; ``model`` is a LangChain model name, such as "openai:gpt-5.2"."""

    # The tools are closures, so they call the OrderDesk passed in here.
    @tool
    def look_up_order(order_id: str) -> str:
        """Look up one order by its order ID and return its items and status."""
        return order_desk.describe_order(order_id)

    @tool
    def cancel_order(order_id: str, reason: str) -> str:
        """Cancel one order that has not shipped yet."""
        return order_desk.cancel_order(order_id, reason)

    # Only a Worker calls the factory, so a Client process needs no provider key.
    def build_order_support_agent() -> Pregel[Any, Any, Any, Any]:
        return create_agent(
            model,
            tools=[look_up_order, cancel_order],
            system_prompt=ORDER_SUPPORT_INSTRUCTIONS,
            middleware=[DurableToolCallMiddleware()],
        )

    return langchain_agent_session_flow(
        build_order_support_agent,
        flow_type=LANGCHAIN_ORDER_SUPPORT_FLOW_TYPE,
        tool_sources=build_slack_tool_sources(slack_connection),
        tool_policies={
            "look_up_order": ToolPolicy(ToolEffect.READ),
            "cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE),
        },
    )

例子: examples/python/agent-harnesses/agent_harness_examples/langchain_order_support.py

Worker

一个 Worker 为 example 中所有 harness 的 session Flow 提供服务。Agent session 的 Step 是 async 的，因此 Python Registry 允许 async handler。每个启动 session 或调用其 RPC 的进程，都要构建同一个 session Flow。

app.pyPython

1234567891011121314151617181920212223242526272829303132

        # Agent session Steps are async, so the Registry allows async handlers.
        self.registry = Registry(
            tuple(example.session_flow for example in self.examples), allow_async_handlers=True
        )
        config.blob_cache_dir.mkdir(parents=True, exist_ok=True)
        self.blob_cache = open_blob_cache(BlobCacheConfig(str(config.blob_cache_dir), 1 << 30))
        self.worker = AsyncWorker(
            self.registry,
            self.blob_cache,
            WorkerOptions(
                server_address=config.server_address,
                pull=PullWorkerOptions(max_concurrent_tasks=MAX_CONCURRENT_TASKS),
            ),
        )
        self.client = AsyncClient(
            self.registry, self.blob_cache, ClientOptions(server_address=config.server_address)
        )

# ...

def _new_langchain_example(
    config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None
) -> AgentHarnessExample:
    order_desk = OrderDesk()
    model = _langchain_model(config.langchain_model)
    flow = build_langchain_order_support_flow(order_desk, model, slack_connection)
    return AgentHarnessExample("langchain", flow, order_desk)

# ...

def _langchain_model(model: str) -> str | BaseChatModel:
    return LangChainDemoChatModel() if is_demo_model(model) else model

例子: examples/python/agent-harnesses/agent_harness_examples/app.py

Session route

每个 harness 都提供这同一组 route。这些 route 和 session token 的说明见 Session route。

http_routes.pyPython

1234567891011121314151617181920212223242526272829303132333435363738394041

    @blueprint.post("/sessions")
    async def start_session() -> Response:
        # The server picks the session ID, so a caller cannot claim an existing session.
        session_id = f"{harness_name}-order-support-{uuid.uuid4()}"
        await sessions.start_session(session_id)
        session_token = sign_session_id(app_state.config.session_key, session_id)
        return jsonify(session_id=session_id, session_token=session_token)

    @blueprint.post("/messages")
    async def send_user_message() -> Response:
        body, session_id = await session_request()
        accepted = await sessions.send_user_message(session_id, _required_string(body, "text"))
        return jsonify(accepted=accepted)

    @blueprint.post("/interrupts/resolve")
    async def resolve_interrupt() -> Response:
        body, session_id = await session_request()
        interrupt_id = _required_string(body, "interrupt_id")
        if not isinstance(approved := body.get("approved"), bool):
            raise BadRequest("approved must be a boolean")
        rejection_message = body.get("rejection_message")
        if rejection_message is not None and not isinstance(rejection_message, str):
            raise BadRequest("rejection_message must be a string")
        accepted = await sessions.resolve_interrupt(
            session_id, interrupt_id, approved=approved, rejection_message=rejection_message
        )
        return jsonify(accepted=accepted)

    @blueprint.get("/snapshot")
    async def get_agent_snapshot() -> Response:
        session_id = request.args.get("session_id", "")
        if not session_id:
            raise BadRequest("session_id is required")
        authorize_session(session_id)
        try:
            snapshot = await read_agent_event_stream_snapshot(
                app_state.client, example.session_flow, session_id
            )
        except FlowNotActiveOrNotFoundError as error:
            raise NotFound("the agent session is missing or no longer retained") from error
        return jsonify(snapshot.to_json())

例子: examples/python/agent-harnesses/agent_harness_examples/http_routes.py

在本地运行

启动 Dex 与 example

在第一个终端中，先渲染 session graph，再启动 Dex。Dex Web 启动时读取这个 graph，用它显示 session 和审批控件。

$$$$$

cd examples/python/agent-harnesses
uv sync --locked
mkdir -p definitions
uv run --frozen dex-agents-session-graph --flow-type LangChainOrderSupport \
  --permission order-support.operate --out definitions/langchain-langgraph.json
dexcli dev --flow-rendering-dir ./definitions

例子: examples/python/agent-harnesses/README.md

dexcli dev 会一直在前台运行。在第二个终端中启动 example：

$$

cd examples/python/agent-harnesses
uv run --frozen python -m agent_harness_examples

例子: examples/python/agent-harnesses/README.md

应用监听 127.0.0.1:8081。Dex Web 位于 127.0.0.1:8802。

没有设置 DEX_LANGCHAIN_MODEL 时，Agent 使用 mock/order-support，这是一个不需要 key 的本地 demo model。收到 "look up order 1001" 时它调用 look_up_order，收到 "cancel order 1002" 时它调用 cancel_order。

变量	默认值	含义
DEX_LANGCHAIN_MODEL	mock/order-support	LangChain 模型名称，例如 openai:gpt-5.2。
OPENAI_API_KEY	无	Provider key，使用 OpenAI 模型时需要。
DEX_AGENT_SESSION_KEY	随机 key	签发 session token 的 key。

调用 session

启动一个 session，让 Agent 取消订单 1002，然后批准这次调用：

$$$$$$$

base=http://127.0.0.1:8081/products/agent-harnesses/langchain
session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' -d '{}')
id=$(echo "$session" | jq -r .session_id)
token=$(echo "$session" | jq -r .session_token)
curl -s -X POST "$base/messages" -H 'Content-Type: application/json' \
  -H "X-Agent-Session-Token: $token" -d "{\"session_id\": \"$id\", \"text\": \"Cancel order 1002\"}"
until interrupt=$(curl -s "$base/snapshot?session_id=$id" -H "X-Agent-Session-Token: $token" \
  | jq -re '.snapshot.pending_interrupts[0].interrupt_id'); do sleep 1; done
curl -s -X POST "$base/interrupts/resolve" -H 'Content-Type: application/json' \
  -H "X-Agent-Session-Token: $token" \
  -d "{\"session_id\": \"$id\", \"interrupt_id\": \"$interrupt\", \"approved\": true}"

例子: examples/python/agent-harnesses/README.md

循环会反复读取 snapshot，直到 Agent 请求运行 cancel_order。这次调用先等待审批，批准后只运行一次。再读一次 snapshot 即可看到 Agent 的回答。

在 Dex Web 中审批

也可以不执行最后一条命令，改在 Dex Web 中审批，步骤见 在 Dex Web 中审批。

连接 Connector

Agent 可以把 Connector 的 operation 当作 tool 调用，见 把 Connector 用作 tool。这个 example 加入了 Slack tool slack_post_thread_reply，用来在 Slack thread 中回复。

提供 tool

安装 dex-agents 的 connectors extra，以及 Connector 包，例如 dex-connector-slack。安装命令见 Connector tools。

Worker 从项目配置中打开 Slack connection support-slack。connector_tool 把 operation 变成 tool，由 ConnectorToolSource 持有。没有这个 connection 时，agent 不会得到 Slack tool。

slack_connector_tools.pyPython

1234567891011121314151617181920212223242526272829303132

def build_slack_tool_sources(connection: slack.Connection | None) -> list[ToolSource]:
    """Returns the Slack tool source, or no source when Slack is not configured."""
    if connection is None:
        return []
    return [
        ConnectorToolSource([
            # The manifest does not ask for approval; this agent asks before every post.
            connector_tool(
                connection.post_thread_reply(), connection.reference, requires_approval=True
            ),
        ]),
    ]

# ...

async def open_slack_project_connection(
    environment: Mapping[str, str] = os.environ,
) -> SlackProjectConnection | None:
    # ...
    # An incomplete or invalid configuration raises here, so the Worker fails at startup.
    try:
        project = await load_from_environment(environment)
    except FileNotFoundError:
        _log_slack_not_set_up("the connector configuration file does not exist")
        return None
    # ...
    try:
        connection = slack.new_project_connection(project, SLACK_CONNECTION_NAME)
    except BaseException:
        await project.close()
        raise
    return SlackProjectConnection(project, connection)

例子: examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py

Worker 在启动时打开一次 connection，每个 agent 都使用它：

__main__.pyPython

123456789

async def main() -> None:
    config = AgentHarnessExamplesConfig.from_env()
    slack_setup = await open_slack_project_connection()
    try:
        app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None)
        # ...
    finally:
        if slack_setup is not None:
            await slack_setup.aclose()

例子: examples/python/agent-harnesses/agent_harness_examples/__main__.py

Agent 把 tool source 和自己的 tool 一起传入：

langchain_order_support.pyPython

123456

    return langchain_agent_session_flow(
        build_order_support_agent,
        flow_type=LANGCHAIN_ORDER_SUPPORT_FLOW_TYPE,
        tool_sources=build_slack_tool_sources(slack_connection),
        # ...
    )

例子: examples/python/agent-harnesses/agent_harness_examples/langchain_order_support.py

调用如何运行

每次调用 slack_post_thread_reply 都要先经人批准，然后在自己的 CallSingleAttemptTool Step 中只运行一次。原因见 Slack Connector tool。

DurableToolCallMiddleware 把 Connector tool 加到 create_agent agent 的每个模型请求中，排在 agent 自己的 tool 之后。自定义 graph 在它的 model 节点中把 get_current_tool_source_tools() 和自己的 tool 一起 bind，durable_tool_node 会像其他调用一样推迟这些调用。这些 tool 的函数体从不在 graph 内运行。

配置 connection

Worker 从哪里找到 connection support-slack、Slack app 需要什么，见 Connection。

试用时，先设置 DEX_CONNECTOR_CONFIG_FILE 再启动 example。然后让 agent 在 Slack thread 中回复，并给出 channel ID 和 thread timestamp：

Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday.

这次发帖和 cancel_order 一样，会先等待审批。

限制

• 如果 graph 的节点在到达 tool node 之前多次调用模型，这些调用都在同一个 RunModel 中。例如 summarization middleware 或 planner 节点。重试会把它们全部重复一遍。
• 由 provider 运行的 tool 会在 RunModel 重跑时重复执行。
• 历史记录只追加。Graph 修改或删除较早的 message，不会改变已保存的历史记录。
• 只有 message 列表会跨 turn 保留。其他 state key 在每个 turn 都从默认值开始。
• Tool 在 graph 之外运行。它们不能使用 ToolRuntime、InjectedState 或 InjectedStore，也不能返回 graph Command。InjectedToolCallId 可以使用。
• Graph 不能自己抛出 interrupt。

Adapter README 列出了全部限制。Server 的限制参见 Limits for long sessions。

相关内容

• Agent Framework
• Agents reference
• Python adapter README