なるべくミニマム実装

stream とかはやってない

import asyncio
from typing import Any, Dict, List, Optional

from claude_code_sdk import AssistantMessage, ClaudeCodeOptions, TextBlock, query
from langchain_core.callbacks import (
    CallbackManagerForLLMRun,
)
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from pydantic import Field


class ChatClaudeCode(BaseChatModel):
    model: str = Field(default="sonnet")

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        last_message = messages.pop()

        chat_history: List[str] = []
        for message in messages:
            if isinstance(message, SystemMessage):
                chat_history.append(f"<system>{message.text()}</system>")
            elif isinstance(message, HumanMessage):
                chat_history.append(f"<human>{message.text()}</human>")
            elif isinstance(message, AIMessage):
                chat_history.append(f"<assistant>{message.text()}</assistant>")
        system_prompt = f"<chat_history>{''.join(chat_history)}</chat_history>"

        async def _aquery_claude():
            options = ClaudeCodeOptions(model=self.model, system_prompt=system_prompt)
            response_content = ""
            async for message in query(prompt=last_message.text(), options=options):
                if isinstance(message, AssistantMessage):
                    for block in message.content:
                        if isinstance(block, TextBlock):
                            response_content += block.text
            return response_content

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            response_content = loop.run_until_complete(_aquery_claude())
        finally:
            loop.close()

        message = AIMessage(
            content=response_content,
            response_metadata={
                "model_name": self.model,
            },
        )

        return ChatResult(generations=[ChatGeneration(message=message)])

    @property
    def _llm_type(self) -> str:
        return "claude-code"

    @property
    def _identifying_params(self) -> Dict[str, Any]:
        return {
            "model_name": self.model,
        }

Claude Code SDK だと複数メッセージを送信する方法がない ( と思う ) ので、無理やり system prompt にまとめて渡してる ( chat_history の部分 )

~/.claude/projects/<project>/<id>.jsonl を直接編集して会話履歴を捏造することで実現できそうな気はする

LangChain の Custom Chat Model に Claude Code SDK を使うサンプル