LangChain / LangGraph 連携

このガイドでは、AI アプリケーションやエージェントを構築するために、clinvk を LangChain / LangGraph と連携させる方法を説明します。

概要

clinvk の OpenAI 互換エンドポイント（/openai/v1/*）により、OpenAI API 形式をサポートする LangChain などの AI フレームワークとシームレスに連携できます。

flowchart LR
    A["LangChain app"] --> B["ChatOpenAI"]
    B --> C["clinvk /openai/v1"]
    C --> D["Backend CLI (claude/codex/gemini)"]

    style A fill:#e3f2fd,stroke:#1976d2
    style B fill:#fff3e0,stroke:#f57c00
    style C fill:#ffecb3,stroke:#ffa000
    style D fill:#f3e5f5,stroke:#7b1fa2

OpenAI SDK Compatibility

clinvk は OpenAI Python SDK と互換です。

from openai import OpenAI

# Configure clinvk as the base URL
client = OpenAI(
    base_url="http://localhost:8080/openai/v1",
    api_key="not-needed"  # only required if API keys are enabled
)

# Use any backend by setting the model name
response = client.chat.completions.create(
    model="claude",  # Backend name: claude, codex, or gemini
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain this code snippet."}
    ]
)

print(response.choices[0].message.content)

LangChain Integration

Basic Chat Model

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# Initialize with clinvk endpoint
llm = ChatOpenAI(
    base_url="http://localhost:8080/openai/v1",
    model="claude",
    api_key="not-needed",
    temperature=0.7
)

# Simple invocation
messages = [
    SystemMessage(content="You are a code review expert."),
    HumanMessage(content="Review this function for bugs.")
]

response = llm.invoke(messages)
print(response.content)

Using with Chains

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Setup
llm = ChatOpenAI(
    base_url="http://localhost:8080/openai/v1",
    model="claude",
    api_key="not-needed"
)

# Create a chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a code documentation expert."),
    ("user", "Generate documentation for this code:\n{code}")
])

chain = prompt | llm | StrOutputParser()

# Execute
result = chain.invoke({"code": "def hello(): return 'world'"})
print(result)

Multiple Backends in One Application

from langchain_openai import ChatOpenAI

# Create multiple LLM instances for different backends
claude_llm = ChatOpenAI(
    base_url="http://localhost:8080/openai/v1",
    model="claude",
    api_key="not-needed"
)

codex_llm = ChatOpenAI(
    base_url="http://localhost:8080/openai/v1",
    model="codex",
    api_key="not-needed"
)

gemini_llm = ChatOpenAI(
    base_url="http://localhost:8080/openai/v1",
    model="gemini",
    api_key="not-needed"
)

# Use different backends for different tasks
architecture_review = claude_llm.invoke("Review the architecture...")
code_generation = codex_llm.invoke("Generate a function that...")
data_analysis = gemini_llm.invoke("Analyze this dataset...")

LangGraph Agent Integration

Basic Agent with Tools

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.prebuilt import create_react_agent
import httpx

# Define a tool that uses clinvk's parallel execution
@tool
def parallel_code_review(code: str) -> dict:
    """Review code using multiple AI backends in parallel."""
    response = httpx.post(
        "http://localhost:8080/api/v1/parallel",
        json={
            "tasks": [
                {
                    "backend": "claude",
                    "prompt": f"Review architecture and design:\n{code}"
                },
                {
                    "backend": "codex",
                    "prompt": f"Review for performance issues:\n{code}"
                },
                {
                    "backend": "gemini",
                    "prompt": f"Review for security vulnerabilities:\n{code}"
                }
            ]
        },
        timeout=120
    )
    return response.json()

@tool
def chain_documentation(code: str) -> str:
    """Generate documentation through a multi-step pipeline."""
    response = httpx.post(
        "http://localhost:8080/api/v1/chain",
        json={
            "steps": [
                {
                    "name": "analyze",
                    "backend": "claude",
                    "prompt": f"Analyze this code structure:\n{code}"
                },
                {
                    "name": "document",
                    "backend": "codex",
                    "prompt": "Generate API docs based on: {{previous}}"
                }
            ]
        },
        timeout=120
    )
    results = response.json()
    return results["results"][-1]["output"]

# Create the agent
llm = ChatOpenAI(
    base_url="http://localhost:8080/openai/v1",
    model="claude",
    api_key="not-needed"
)

agent = create_react_agent(llm, [parallel_code_review, chain_documentation])

# Run the agent
result = agent.invoke({
    "messages": [{"role": "user", "content": "Review this code: def add(a, b): return a + b"}]
})

Custom Agent Graph

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class AgentState(TypedDict):
    code: str
    architecture_review: str
    performance_review: str
    security_review: str
    final_report: str

def review_architecture(state: AgentState) -> AgentState:
    llm = ChatOpenAI(
        base_url="http://localhost:8080/openai/v1",
        model="claude",
        api_key="not-needed"
    )
    result = llm.invoke(f"Review architecture:\n{state['code']}")
    return {"architecture_review": result.content}

def review_performance(state: AgentState) -> AgentState:
    llm = ChatOpenAI(
        base_url="http://localhost:8080/openai/v1",
        model="codex",
        api_key="not-needed"
    )
    result = llm.invoke(f"Review performance:\n{state['code']}")
    return {"performance_review": result.content}

def review_security(state: AgentState) -> AgentState:
    llm = ChatOpenAI(
        base_url="http://localhost:8080/openai/v1",
        model="gemini",
        api_key="not-needed"
    )
    result = llm.invoke(f"Review security:\n{state['code']}")
    return {"security_review": result.content}

def generate_report(state: AgentState) -> AgentState:
    llm = ChatOpenAI(
        base_url="http://localhost:8080/openai/v1",
        model="claude",
        api_key="not-needed"
    )
    prompt = f"""Generate a final report based on these reviews:

Architecture: {state['architecture_review']}
Performance: {state['performance_review']}
Security: {state['security_review']}
"""
    result = llm.invoke(prompt)
    return {"final_report": result.content}

# Build the graph
workflow = StateGraph(AgentState)

workflow.add_node("architecture", review_architecture)
workflow.add_node("performance", review_performance)
workflow.add_node("security", review_security)
workflow.add_node("report", generate_report)

workflow.set_entry_point("architecture")
workflow.add_edge("architecture", "performance")
workflow.add_edge("performance", "security")
workflow.add_edge("security", "report")
workflow.add_edge("report", END)

app = workflow.compile()

# Run
result = app.invoke({"code": "def process(data): return data * 2"})
print(result["final_report"])

Streaming Responses

OpenAI SDK Streaming

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8080/openai/v1",
    api_key="not-needed"
)

# Stream the response
stream = client.chat.completions.create(
    model="claude",
    messages=[{"role": "user", "content": "Write a long explanation..."}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

LangChain Streaming

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    base_url="http://localhost:8080/openai/v1",
    model="claude",
    api_key="not-needed",
    streaming=True
)

for chunk in llm.stream([HumanMessage(content="Explain recursion...")]):
    print(chunk.content, end="", flush=True)

Async Support

import asyncio
from openai import AsyncOpenAI

async def main():
    client = AsyncOpenAI(
        base_url="http://localhost:8080/openai/v1",
        api_key="not-needed"
    )

    # Async completion
    response = await client.chat.completions.create(
        model="claude",
        messages=[{"role": "user", "content": "Hello!"}]
    )
    print(response.choices[0].message.content)

    # Async streaming
    stream = await client.chat.completions.create(
        model="claude",
        messages=[{"role": "user", "content": "Tell me a story..."}],
        stream=True
    )

    async for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="")

asyncio.run(main())

Error Handling

from openai import OpenAI, APIError, APIConnectionError

client = OpenAI(
    base_url="http://localhost:8080/openai/v1",
    api_key="not-needed"
)

try:
    response = client.chat.completions.create(
        model="claude",
        messages=[{"role": "user", "content": "Hello"}]
    )
except APIConnectionError:
    print("Could not connect to clinvk server. Is it running?")
except APIError as e:
    print(f"API error: {e.message}")

Best Practices

1. Connection Pooling

import httpx

# Reuse client for multiple requests
client = httpx.Client(
    base_url="http://localhost:8080",
    timeout=120
)

# Use in your application
response = client.post("/api/v1/prompt", json={...})

2. Timeout Configuration

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    base_url="http://localhost:8080/openai/v1",
    model="claude",
    api_key="not-needed",
    request_timeout=120  # 2 minutes for long tasks
)

3. Backend Selection Strategy

タスク	推奨バックエンド	理由
複雑な推論	claude	分析能力が強い
コード生成	codex	コード向けに最適化
データ分析	gemini	構造化データに強い
一般的なタスク	claude	汎用的なデフォルトとして有用

Next Steps

• CI/CD - パイプラインで自動化
• クライアントライブラリ - 他言語のバインディング
• REST API リファレンス - API ドキュメント一式