Conductor 与 LangChain 集成指南：将现有 LangChain Agent 编译为持久化可观测的工作流执行-开发者社区

Conductor 与 LangChain 集成指南：将现有 LangChain Agent 编译为持久化可观测的工作流执行

【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor

导读：本文以 ui-next/src/pages/agent/guides/python/langchain.md 为骨架，讲解如何通过conductor-python[langchain]安装包，将用langchain.agents.create_agent编写的 LangChain Agent 交给 Conductor 的AgentRuntime运行，使其成为一次持久化（durable）、可观测（inspectable）的 Conductor 工作流执行。读完本文，你将掌握环境变量配置、Agent 编写与运行、run/plan/deploy/serve四个生命周期操作，以及把部署后的 LangChain Agent 通过AGENT任务接入父工作流的生产级用法。

1. 前置准备：先连接 Conductor，再配置模型凭证

在运行任何 LangChain Agent 之前，需要保证两件事：Conductor 运行时可以连到你的服务器，服务器可以调用你选择的模型提供商。完整步骤见 Connect to Conductor。

方式一：托管版（Developer Edition）：创建账号与应用，取得 Access Key 后设置：

export CONDUCTOR_SERVER_URL=https://developer.orkescloud.com/api export CONDUCTOR_AUTH_KEY=<your-access-key> export CONDUCTOR_AUTH_SECRET=<your-access-secret>

方式二：本地服务器：需要 Java 21+ 与 Node.js，通过 CLI 启动本地开发服务器：

conductor server start export CONDUCTOR_SERVER_URL=http://localhost:8080/api

模型凭证：Agent 的模型调用发生在服务器侧。托管版在 Integrations 中添加 AI/LLM 集成；本地服务器则在启动前导出提供商 API Key，例如：

export OPENAI_API_KEY=<your-openai-api-key> conductor server start

Conductor 内置的 LLM 提供商包括 Anthropic、OpenAI、Azure OpenAI、Google Gemini、AWS Bedrock、Mistral、Cohere、HuggingFace、Ollama、Perplexity、Grok (xAI) 等，完整支持矩阵见 LLM orchestration。按官方文档的建议，提供商凭证应放在 Agent Worker 的环境变量或密钥系统中，不要写进工作流输入（参见 Your First Agent）。

2. 安装：`conductor-python[langchain]`扩展包

原文档给出的安装命令为：

python -m pip install 'conductor-python[langchain]'

[langchain]是 extras 标记，安装时会一并拉取 LangChain 桥接所需的依赖。同一系列的其他 extras 还包括conductor-python[langgraph]（LangGraph 支持）与conductor-python[adk]（Google ADK 支持），详见 Framework Agents 参考页。如果只想运行 Conductor 原生Agent，基础包pip install conductor-python即可（参见 Your First Agent）。

3. 配置环境变量

安装完成后，配置 Conductor 服务器地址与认证信息，并指定默认的 LLM 模型：

export CONDUCTOR_SERVER_URL={{CONDUCTOR_SERVER_URL}} # 对于需要认证的 Conductor 服务器： # export CONDUCTOR_AUTH_KEY=<YOUR_AUTH_KEY> # export CONDUCTOR_AUTH_SECRET=<YOUR_AUTH_SECRET> export CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini

CONDUCTOR_SERVER_URL：Conductor API 地址。本地服务器为http://localhost:8080/api。
CONDUCTOR_AUTH_KEY/CONDUCTOR_AUTH_SECRET：访问控制凭证，仅在启用认证的服务器上必需。
CONDUCTOR_AGENT_LLM_MODEL：Agent 的默认模型，使用provider/model形式。create_agent的第一个参数（如"openai:gpt-4o-mini"）也可在代码中显式指定模型。

4. 编写并运行第一个 LangChain Agent

原文档给出了最小可运行示例，保存为langchain_agent.py：

from conductor.ai.agents import AgentRuntime from langchain.agents import create_agent agent = create_agent( "openai:gpt-4o-mini", tools=[], system_prompt="You are a concise assistant.", ) with AgentRuntime() as runtime: result = runtime.run(agent, "Explain durable execution in one sentence.") result.print_result()

运行：

python langchain_agent.py

代码拆解：

create_agent(model, tools, system_prompt)：来自langchain.agents，定义 Agent 的模型、工具集与系统提示词。它是 LangChain 侧的创作面（authoring surface）。
AgentRuntime：来自conductor.ai.agents，是 Conductor SDK 的运行时入口。用with语句进入上下文，退出时自动释放资源。
runtime.run(agent, input)：把 LangChain Agent 对象编译为 Conductor 工作流图并立即执行一次，返回可打印的结果对象。
result.print_result()：打印最终输出文本。

4.1 带工具的 LangChain Agent

工具是框架侧能力的关键。使用langchain_core.tools.tool装饰器定义工具函数，并传入tools列表（参见 Framework Agents 快速上手）：

from conductor.ai.agents import AgentRuntime from langchain.agents import create_agent from langchain_core.tools import tool @tool def check_token() -> str: """Check a token.""" return "available" agent = create_agent("openai:gpt-4o-mini", tools=[check_token], system_prompt="You are a helpful assistant.") with AgentRuntime() as runtime: result = runtime.run(agent, "Is the token set?") result.print_result()

关键机制：工具函数（如check_token）不是运行在 Conductor 服务器上，而是在 SDK 侧作为本地 Worker 进程执行。因此生产环境中必须保持承载工具的 Worker 进程存活，Agent 才能调用到这些工具。

5. 深入：从 LangChain Agent 到工作流图的四个生命周期操作

在原文档的最小示例之上，Python SDK 把“从框架对象到可复用工作流步骤”的过程收敛为同一个运行时的四个调用（详见 Framework Agents）：

from conductor.ai.agents import AgentRuntime from langchain.agents import create_agent from langchain_core.tools import tool @tool def check_token() -> str: """Check a token.""" return "available" agent = create_agent("openai:gpt-4o-mini", tools=[check_token], system_prompt="You are a helpful assistant.") with AgentRuntime() as runtime: runtime.run(agent, "Is the token set?") # develop: 编译并执行一次 runtime.plan(agent) # CI: 检查编译后的工作流图 runtime.deploy(agent) # release: 注册但不执行 runtime.serve(agent) # operate: 运行工具 Worker 并阻塞

操作	阶段	作用
`run`	开发迭代	编译并执行一次，生成可见于 UI 的持久化执行记录
`plan`	CI	检查 Agent 将被编译成的工作流图，不产生执行
`deploy`	发布	在服务器上注册为具名、带版本的 Conductor Agent，不执行
`serve`	运维	启动工具 Worker 进程并阻塞，保持工具可调用

其中serve()是阻塞调用，生产环境中应放入独立的长期存活 Worker 进程；deploy()则放在 CI/CD 流水线中执行。其余框架（OpenAI Agents、LangGraph、Google ADK）走同一套生命周期模型。

5.1 服务器端开关：AI 集成必须启用

部署或调用 Conductor Agent 之前，服务器必须开启 AI 集成开关（见 Conductor Agents）：

conductor.integrations.ai.enabled=true

当该属性为false或缺失时，已部署 Agent 的控制面以及agentType: "conductor"执行模式均不可用。

6. 把部署后的 LangChain Agent 接入父工作流：`AGENT`任务

部署完成后，LangChain 就不再是调用方必须了解的细节——父工作流通过AGENT任务按名称调用已部署的 Agent（见 Conductor Agents）：

{ "name": "run_agent", "taskReferenceName": "run_agent_ref", "type": "AGENT", "inputParameters": { "agentType": "conductor", "name": "<deployed-agent-name>", "prompt": "${workflow.input.prompt}" } }

重要语义：agentType选择的是执行模式，而不是创作框架。agentType: "a2a"（默认）调用远程 A2A 端点；agentType: "conductor"按name运行部署在本机的 Conductor Agent。LangChain、LangGraph、ADK 等只是 SDK 创作路径，不是agentType的取值。

参数说明：

name：已部署 Agent 名称，首次调用时必填。
prompt：传给 Agent 的输入文本，首次调用时必填。
version：可选，固定部署版本；省略则使用最新版本。
其他可选参数：sessionId、runId、context、media、model、timeoutSeconds、idempotencyKey等。运行时在未提供时会生成重启稳定的幂等键。

AGENT任务会写入executionId、agentName、state、text以及完成时的结构化output。state采用归一化的 A2A 生命周期值：working、input-required、completed、failed、canceled，并映射到 Conductor 任务状态（IN_PROGRESS/COMPLETED/FAILED/CANCELED）。maxDurationSeconds（默认 86400 秒）约束整次运行，maxPollFailures（默认 30）约束连续瞬态轮询失败，两者都会终态失败任务并尽力取消子执行。

7. 实战案例：LangChain 编写的授权调查 Agent

仓库 cookbook 中提供了一个完整的 LangChain 实战案例 langchain-entitlement-investigator：用 LangChain 编写授权调查（entitlement investigation）Agent，通过 Conductor SDK 部署，再以持久化能力被父工作流调用。

Agent 定义（LangChain 侧，工具由配套部署脚本通过 MCP 提供）：

from langchain.agents import create_agent # 配套部署脚本提供两个真实的 MCP 适配器 agent = create_agent( "openai:gpt-4o", tools=[list_mcp_testkit_tools, call_mcp_testkit_tool], system_prompt="Investigate entitlements from MCP evidence; recommend only.", )

配套脚本 deploy_local_cookbook_agents.py 展示了部署细节：它用conductor.ai.agents中的AgentRuntime、mcp_tool、tool、RegexGuardrail、Position、OnFail等组件，把 LangChain Agent 与本地 MCP Testkit 服务器（http://127.0.0.1:3001/mcp）绑定，并演示了在工具上叠加守卫（guardrail）与审批边界的写法：

@tool(guardrails=[no_payment_card_data], approval_required=True) def request_notification(destination: str, summary: str) -> dict[str, str]: """Request an approved notification; replace with an idempotent integration.""" return {"status": "approved-notification-requested", "destination": destination}

部署一次并保持 Worker 运行：

python3 deploy_local_cookbook_agents.py deploy python3 deploy_local_cookbook_agents.py serve

配套的工作流定义 langchain-entitlement-investigator.json 展示了“稳定契约、框架无关”的调用方式：

{ "name": "langchain_entitlement_investigator", "version": 1, "schemaVersion": 2, "timeoutSeconds": 300, "timeoutPolicy": "TIME_OUT_WF", "inputParameters": ["customerId", "question"], "tasks": [ { "name": "investigate_entitlement", "taskReferenceName": "investigate_entitlement", "type": "AGENT", "inputParameters": { "agentType": "conductor", "name": "langchain-entitlement-investigator", "prompt": "Customer ${workflow.input.customerId}: ${workflow.input.question}" } } ], "outputParameters": { "investigation": "${investigate_entitlement.output.output}", "executionId": "${investigate_entitlement.output.executionId}" } }

注册并运行：

conductor workflow create langchain-entitlement-investigator.json conductor workflow start -w langchain_entitlement_investigator --sync -i '{"customerId":"C-123","question":"Which plan features are enabled?"}'

该案例印证了“只读工具 + 人为审批边界”的生产模式：给 Agent 只读的授权查询工具，任何变更动作都必须经过人审批准的外部动作通道。仓库中另有四个框架无关的工作流集成示例可供对照，见 ai/examples 下的31-conductor-agent-basic.json（单步复用）、32-conductor-agent-human-in-loop.json（WAITING → HUMAN → 用 executionId 恢复）、33-conductor-agent-multi-agent.json（FORK_JOIN/JOIN中的并行专业 Agent）、34-conductor-agent-cancel.json（父图取消传播）。

8. 生产化注意事项

原文档及其配套 cookbook 明确给出的生产要点，整理如下：

agentType是conductor，不是langchain。执行协议由 Conductor 运行时承载，不会因为创作框架是 LangChain 而改变。
部署后的 Agent 内约束 token 与工具调用。Agent 循环（LLM 调用、工具调用、等待、重试、分支）实际运行在部署后的工作流图中，因此要在那里设置限制，而不是依赖外层兜底。
传递文档引用而非载荷。输入中传文档引用（引用/ID），不要塞入大段载荷。
用“客户 ID + 请求 ID”对重复运行做对账。Agent 运行可能因重试而重复，幂等与对账策略要提前设计。
升级包版本前核对 SDK 源码。Framework-agent API 仍在演进，应以 SDK 与仓库文档为准。
不要把副作用交给 Agent 自行拍板。参考 Production agent architecture 的父工作流模式：父工作流验证请求 → 选择执行边界（原生任务 / 已部署 Conductor Agent / 远程 A2A）→ 验证返回结果 → 审批后写入或失败补偿。任何不可逆操作都不能仅凭 Agent 的输出直接执行。

9. 验证与故障排查

运行python langchain_agent.py后：

验证输出：检查终端打印的结果文本是否符合预期。
在 Conductor UI 中核对执行：每次run都会生成一条持久化执行记录，可在 UI 中查看任务时间线、输入与输出，确认 Agent 的每一次 LLM 调用、工具调用、等待与分支都可观测。
失败排查顺序：先确认运行时服务器 URL（CONDUCTOR_SERVER_URL）、框架包版本、提供商凭证（模型调用发生在服务器侧）；然后检查执行记录中的失败任务，再决定重试。
重试纪律：对于可能已产生外部副作用的 Agent 动作，在幂等性与恢复策略明确之前，不要盲目重试（见 Framework Agents 快速上手的 Verify and recover 一节）。

10. 进阶阅读

Framework Agents：LangChain / LangGraph / ADK / Vercel AI SDK 的参考矩阵与生命周期详解
Your First Agent：不依赖框架、用 Conductor 原生Agent起步
Conductor Agents：已部署 Agent 的运行契约、恢复、取消与输出定义
LLM orchestration：服务器侧 LLM 提供商与原生 AI 任务
Production agent architecture：生产级父工作流参考架构
设计模式与完整可运行示例：见 Agent Cookbook 索引，其中 LangChain investigator 与本文主题直接对应

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Conductor 与 LangChain 集成指南：将现有 LangChain Agent 编译为持久化可观测的工作流执行