LangChain 和 LangGraph

把 OpenViking 接入你的 LangChain 或 LangGraph Agent 作为上下文后端。独立集成包提供 retriever、chat history、context wrapper、agent tools、LangGraph store 和 middleware， 可连接 HTTP 服务或嵌入式 OpenViking。

安装

pip install langchain-openviking                 # LangChain 适配器
pip install "langchain-openviking[langgraph]"    # LangGraph middleware

该集成独立于 OpenViking server 发布。为兼容现有应用，完整包仍会把旧的 openviking.integrations.langchain 导入路径转发到 langchain-openviking。

连接

from langchain_openviking import create_openviking_tools

tools = create_openviking_tools(
    url="http://localhost:1933",
    api_key="...",
    profile="agent",
)

同时省略 url 和 path 时，适配器会使用 OpenViking CLI 配置中的 HTTP 连接信息。传入 path 时，通过 OpenViking 同步 client 使用嵌入式 workspace，该模式还需要安装完整的 openviking 包。Embedding 和 VLM 在 OpenViking 侧配置，不在你的应用中。

异步应用

Retriever、context wrapper、chat history、session recorder 和 LangGraph middleware 都支持原生异步路径。通过 URL 配置时，适配器会自动创建异步 OpenViking HTTP client：

docs = await retriever.ainvoke("用户之前做了什么决定？")
result = await chain.ainvoke(
    {"messages": [...]},
    config={"configurable": {"session_id": "support-thread-1"}},
)

异步适配器支持三种 client 模式：

配置	异步接口	所有权
client= 或 async_client=	原样返回注入的 client	调用方
url=，或同时省略 url 和 path	每个 event loop 一个支持恢复的 HTTP handle	Adapter
path=	在 worker thread 中调用同步嵌入式 client	Adapter

长期运行的应用可以初始化一个由调用方管理的异步 client，并在同一 event loop 内的多个适配器之间复用：

from openviking_sdk import AsyncHTTPClient
from langchain_openviking import OpenVikingRetriever

client = AsyncHTTPClient(url="http://localhost:1933", api_key="...")
await client.initialize()
try:
    retriever = OpenVikingRetriever(async_client=client)
    docs = await retriever.ainvoke("部署决定")
finally:
    await client.close()

注入的异步 client 会绑定到初始化它的 event loop。不要跨 event loop 共享同一个 注入异步 client；应为每个 loop 分别创建并管理 client。注入的同步 client 仍可安全地 用于异步 adapter 方法，因为调用会在 worker thread 中执行。

path= 嵌入式 adapter 使用同步 fallback 是有意设计：SyncOpenViking 会让有状态的 嵌入式引擎保持在 OpenViking 的共享后台 loop 上，同时不阻塞应用 event loop。若要使用 原生嵌入式异步方法，请自行创建并初始化 AsyncOpenViking，通过 async_client= 注入， 在同一个 event loop 中使用，并由调用方自行关闭。每个进程同时只能运行一个嵌入式 workspace；切换 workspace 前应先关闭或 reset 当前 client。

OpenVikingChatMessageHistory 提供 aget_messages()、aadd_messages() 和 aclear()；OpenVikingSessionRecorder 提供 arecord()、aflush() 和 aclose()。异步 LangGraph 运行会自动选择 awrap_model_call() 和 aafter_agent()。同一 adapter 首次被并发调用时，每个 event loop 只会创建一个内部 HTTP client。通过 with_openviking_context() 执行的每次 runnable 调用都独立持有 本次写入所需的 history 快照、peer 身份和召回上下文引用。因此，同一 session 的调用 可以并发执行，不会因为 退出时再次读取实时 history 而丢失消息；未消费完的 stream 也不会占用 session 级 lifecycle lock。只有最终的 append-and-commit 步骤会被串行化：async 写入在每个 event loop 内按 session 串行执行，sync 写入则会跨线程按 session 串行执行。

如果 recorder 已确认部分写入后任务被取消，arecord() 会重新抛出原始 asyncio.CancelledError。可将该异常或外层的 asyncio.TimeoutError 传给 get_openviking_cancellation_progress()，在重试前读取已确认写入的消息前缀或待 commit 状态，避免重复写入。保留原始取消异常也会保留 asyncio.wait_for() 和 asyncio.timeout() 的标准超时行为。

Adapter 不会关闭调用方注入的 client。对于 adapter 自行创建的 client，应按实际使用的组件调用 await retriever.aclose()、await assembler.aclose()、 await middleware.aclose()、await history.aclose() 或 await recorder.aclose()。如果 async 操作完成后误调用同步 recorder.close()，该方法会抛出异常并保持 recorder 可用，以便后续 await recorder.aclose() 仍能释放全部资源。 如果条件允许，应在关闭 event loop 前关闭 HTTP-backed adapter；原始 loop 已结束后的 清理属于 best-effort。

with_openviking_context() 返回 OpenVikingContextRunnable。它兼容 LangChain 的 RunnableWithMessageHistory，并负责管理其创建的 context 和 recording adapter。 它会在多次调用之间复用按 event loop 隔离的 client，同时继续隔离每次调用的 history、 peer 身份和召回引用。推荐使用托管生命周期：

async with with_openviking_context(runnable, url="http://localhost:1933") as chain:
    result = await chain.ainvoke(
        {"messages": [...]},
        config={"configurable": {"session_id": "support-thread-1"}},
    )

同步调用使用 with ...，也可以显式调用 close() 或 await aclose()。不要在正在运行的 event loop 中调用 close()，此时应使用 aclose()。注入的 client 仍由调用方管理。

LCEL 组合会返回普通的 RunnableSequence，不会暴露 OpenViking 的 close 方法。应保留 托管 wrapper，并在其生命周期内完成组合：

async with with_openviking_context(runnable, url="http://localhost:1933") as managed:
    chain = managed | another_step
    result = await chain.ainvoke(...)

Peer 身份

传入 actor_peer_id 可以在文件系统和检索操作中过滤当前用户的 peer 集合。session message capture 仍可使用 peer_id 表达每条消息的说话人归属。

retriever = OpenVikingRetriever(
    url="http://localhost:1933",
    actor_peer_id="assistant-a",
)

chain = with_openviking_context(
    runnable,
    session_id="support-thread-1",
    actor_peer_id="assistant-a",
)

动态运行时，with_openviking_context() 默认仍会读取 config["configurable"]["peer_id"]，用于 captured message 的归属：

chain.invoke(
    {"messages": [...]},
    config={"configurable": {"session_id": "support-thread-1", "peer_id": "assistant-a"}},
)

并发 Agent 的运行时 Actor Peer

OpenVikingContextMiddleware 可以在复用绑定凭证的 HTTP client 时，从每次 LangGraph 运行中解析当前 actor peer：

from langchain_openviking import OpenVikingContextMiddleware


def resolve_actor_peer(_state, runtime):
    context = runtime.context or {}
    return context.get("actor_peer_id")


middleware = OpenVikingContextMiddleware(
    url="http://localhost:1933",
    api_key="user-api-key",
    actor_peer_resolver=resolve_actor_peer,
)

解析出的 actor peer 会作用于召回和捕获期间发出的 OpenViking HTTP 请求。并发运行 互相隔离，middleware 的捕获进度也会按 actor peer、session 和 message peer 共同 隔离。OpenViking 的 Session 接口仍然以 user 为作用域，不会使用 actor-peer header 标记消息归属；如果捕获的消息也需要归属于同一个逻辑 peer，应同时设置 peer_id_resolver。拥有独立历史的不同 peer 也应解析为不同的 session ID。未传入 actor_peer_resolver 时，现有固定 client 行为保持不变。

该 resolver 不能改变 OpenViking account 或 user；这些身份继续由 API Key 或 OAuth 凭证决定。因此，多用户应用必须先选择绑定对应用户凭证的 client，再调用 middleware。 Actor peer 只能从已经认证、由服务端控制的 runtime 字段中解析；不要信任 model state 或客户端可控的 configurable 值。运行时 actor-peer 解析仅支持 HTTP-backed middleware，不支持 embedded path= client。注入的自定义 client 必须设置 supports_request_actor_peer = True，并遵循 openviking_sdk 的 actor-peer 作用域。在已有环境中启用该能力前，应同时升级 openviking-sdk 和 openviking。

选哪个适配器？

我想…	用这个
为 RAG 检索相关上下文	OpenVikingRetriever
包装 runnable，自动召回 + 捕获 + 按策略 commit	with_openviking_context()
给 agent 暴露显式记忆工具	create_openviking_tools()
存储跨线程的持久化状态	OpenVikingStore
在 LangGraph 中以 middleware 注入上下文	OpenVikingContextMiddleware
用 OpenViking 存储 LangChain 聊天记录	OpenVikingChatMessageHistory
在自定义生命周期中记录调用方选定的 LangChain 消息	OpenVikingSessionRecorder

快速示例

Retriever

from langchain_openviking import OpenVikingRetriever

retriever = OpenVikingRetriever(url="http://localhost:1933", api_key="...")
docs = retriever.invoke("用户之前对部署方案做了什么决定？")

Context backend

from langchain_core.messages import AIMessage
from langchain_core.runnables import RunnableLambda
from langchain_openviking import with_openviking_context

with with_openviking_context(
    RunnableLambda(lambda msgs: AIMessage(content="...")),
    url="http://localhost:1933",
    api_key="...",
) as chain:
    result = chain.invoke(...)

Agent tools

from langchain_openviking import create_openviking_tools

tools = create_openviking_tools(url="http://localhost:1933", profile="agent")
# 包括：viking_find, viking_search, viking_browse, viking_read,
#       viking_grep, viking_store, viking_add_resource 等

LangGraph store

from langchain_openviking import OpenVikingStore

store = OpenVikingStore(url="http://localhost:1933", api_key="...")
store.put(("users", "ada"), "preferences", {"color": "azure"})
items = store.search(("users",), query="azure", limit=3)

LangGraph middleware

from langchain_openviking import OpenVikingContextMiddleware

middleware = OpenVikingContextMiddleware(
    url="http://localhost:1933",
    api_key="...",
    capture_on_after_agent=True,
)

Session recorder

当应用已经自行管理会话生命周期，只需要复用 OpenViking 持久化能力时，可使用 recorder：

from langchain_openviking import (
    OpenVikingPartialWriteError,
    OpenVikingSessionRecorder,
)

recorder = OpenVikingSessionRecorder(url="http://localhost:1933", api_key="...")
try:
    recorder.record("support-thread-1", messages, peer_id="assistant-a")
except OpenVikingPartialWriteError as exc:
    recorder.record(
        "support-thread-1",
        messages[exc.input_messages_consumed :],
        peer_id="assistant-a",
    )
recorder.flush("support-thread-1")
recorder.close()

record() 只写入调用方传入的消息；它会过滤框架控制消息、按服务端限制分批写入，并应用已配置的 commit 策略。如果后续批次或写入后的 commit 失败，OpenVikingPartialWriteError 会报告已经 确认写入的输入前缀，调用方可仅重试尚未写入的后缀；空后缀会安全地重试待完成的 commit。 传入 context_parts 时，仅在 exc.context_attached 为 false 时重传。flush() 仅在 session 存在待提交内容时强制 commit。close() 后 recorder 不可复用；由调用方注入的 client 仍归调用方管理。

异步生命周期使用对应的 await recorder.arecord(...)、 await recorder.aflush(...) 和 await recorder.aclose()。不要用 recorder.close() 结束异步生命周期。

运行示例

仓库内提供了可直接运行的最小示例，使用内存测试客户端，无需模型凭证：

uv run --project integrations/langchain --extra langgraph python examples/langchain-langgraph/langchain/rag/quick_app.py
uv run --project integrations/langchain --extra langgraph python examples/langchain-langgraph/langchain/context-backend/quick_app.py
uv run --project integrations/langchain --extra langgraph python examples/langchain-langgraph/langchain/message-history/quick_app.py
uv run --project integrations/langchain --extra langgraph python examples/langchain-langgraph/langgraph/agent/quick_app.py
uv run --project integrations/langchain --extra langgraph python examples/langchain-langgraph/langgraph/middleware/quick_app.py

连接真实 OpenViking 服务和 OpenAI 兼容模型的示例见 live LangGraph app。

参见

• examples/langchain-langgraph/ — 上面所有示例的完整源码
• MCP 客户端 — 非 SDK 方式的 MCP 集成