Tools

from langchain.tools import tool

from docsassistant.search import retriever   # pgvector-backed in DocsAssistant


@tool(parse_docstring=True)
def search_docs(query: str, limit: int = 4) -> str:
    """Search the documentation site and return matching sections.

    Use a key phrase from the question, not the whole question.

    Args:
        query: Key phrase to search for, for example "HNSW ef_search".
        limit: Maximum number of sections to return.
    """
    hits = retriever(query, k=limit)
    if not hits:
        return "No matching sections. Try a shorter key phrase."
    return "\n".join(f"{h['page']} :: {h['section']} :: {h['content']}" for h in hits)


print(search_docs.name)
print(search_docs.description)
print(search_docs.invoke({"query": "HNSW"}))

search_docs
Search the documentation site and return matching sections. Use a key phrase from the question, not the whole question.
modules/ROOT/pages/database/vector-rag/ann-indexes.adoc :: HNSW :: HNSW ef_search trades recall for latency.
modules/ROOT/pages/database/vector-rag/postgresql-pgvector.adoc :: Indexes :: pgvector supports HNSW and IVFFlat.

import json

print(json.dumps(search_docs.tool_call_schema.model_json_schema(), indent=1))

{
 "description": "Search the documentation site and return matching sections. Use a key phrase from the question, not the whole question.",
 "properties": {
  "query": {
   "description": "Key phrase to search for, for example \"HNSW ef_search\".",
   "title": "Query",
   "type": "string"
  },
  "limit": {
   "default": 4,
   "description": "Maximum number of sections to return.",
   "title": "Limit",
   "type": "integer"
  }
 },
 "required": [
  "query"
 ],
 "title": "search_docs",
 "type": "object"
}

from typing import Literal

from pydantic import BaseModel, Field


class SearchInput(BaseModel):
    """Input for documentation search."""

    query: str = Field(description="Key phrase to search for")
    component: Literal["vector-rag", "mcp"] = Field(default="vector-rag", description="Documentation area")


@tool("search_docs_v2", args_schema=SearchInput)
def search_docs_v2(query: str, component: str = "vector-rag") -> str:
    """Search one documentation area."""
    return f"{component}: {query}"


print(search_docs_v2.args)
print(search_docs_v2.invoke({"query": "x", "component": "mcp"}))
search_docs_v2.invoke({"query": "x", "component": "nope"})   # raises pydantic ValidationError

{'query': {'description': 'Key phrase to search for', 'title': 'Query', 'type': 'string'}, 'component': {'default': 'vector-rag', 'description': 'Documentation area', 'enum': ['vector-rag', 'mcp'], 'title': 'Component', 'type': 'string'}}
mcp: x
ValidationError

from langchain_ollama import ChatOllama

model = ChatOllama(model="llama3.1:8b", temperature=0)
bound = model.bind_tools([search_docs])

print(type(bound).__name__)
print(bound.kwargs["tools"][0]["function"]["name"])

from langchain_core.messages import AIMessage

reply = AIMessage(
    content="",
    tool_calls=[{"name": "search_docs", "args": {"query": "HNSW"}, "id": "call_1", "type": "tool_call"}],
)
tool_message = search_docs.invoke(reply.tool_calls[0])

print(type(tool_message).__name__, tool_message.tool_call_id, tool_message.name)

ToolMessage call_1 search_docs

from dataclasses import dataclass

from langchain.agents import AgentState, create_agent
from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.types import Command


@dataclass
class DocsContext:
    user_id: str
    component: str


class DocsState(AgentState):
    last_query: str


@tool
def scoped_search(query: str, runtime: ToolRuntime[DocsContext]) -> str:
    """Search the docs of the user's current component."""
    n = len(runtime.state["messages"])
    return f"[{runtime.context.component}] results for {query!r} (messages so far: {n}, user: {runtime.context.user_id})"


@tool
def remember_topic(topic: str, runtime: ToolRuntime[DocsContext, DocsState]) -> Command:
    """Remember the topic the user is interested in, for later sessions."""
    runtime.store.put(("topics", runtime.context.user_id), "latest", {"topic": topic})
    return Command(update={
        "last_query": topic,
        "messages": [ToolMessage(content=f"Noted: {topic}", tool_call_id=runtime.tool_call_id)],
    })


print("schema seen by model:", list(scoped_search.tool_call_schema.model_json_schema()["properties"]))

store = InMemoryStore()
agent = create_agent(
    model,                                   # scripted fake model in the run below
    tools=[scoped_search, remember_topic],
    context_schema=DocsContext,
    state_schema=DocsState,
    store=store,
    checkpointer=InMemorySaver(),
)
out = agent.invoke(
    {"messages": [{"role": "user", "content": "Tell me about HNSW"}]},
    config={"configurable": {"thread_id": "t1"}},
    context=DocsContext(user_id="u1", component="vector-rag"),
)
print("state field:", out["last_query"])
print("store:", store.get(("topics", "u1"), "latest").value)

schema seen by model: ['query']
HumanMessage | Tell me about HNSW
AIMessage |
ToolMessage | [vector-rag] results for 'HNSW' (messages so far: 2, user: u1)
AIMessage |
ToolMessage | Noted: HNSW
AIMessage | Done.
state field: HNSW
store: {'topic': 'HNSW'}

from langchain.tools import tool


@tool(response_format="content_and_artifact")
def search_docs_with_sources(query: str) -> tuple[str, list[dict]]:
    """Search the documentation and return a short summary for the model."""
    hits = retriever(query)
    summary = "\n".join(f"- {h['page']} ({h['section']})" for h in hits) or "No matches."
    return summary, hits        # (content for the model, artifact for the application)


call = {"name": "search_docs_with_sources", "args": {"query": "HNSW"}, "id": "call_1", "type": "tool_call"}
msg = search_docs_with_sources.invoke(call)
print(type(msg).__name__)
print("content :", msg.content.replace("\n", " | "))
print("artifact:", [h["section"] for h in msg.artifact])
print("plain invoke (no tool call):", type(search_docs_with_sources.invoke({"query": "HNSW"})).__name__)

ToolMessage
content : - modules/ROOT/pages/database/vector-rag/ann-indexes.adoc (HNSW) | - modules/ROOT/pages/database/vector-rag/postgresql-pgvector.adoc (Indexes)
artifact: ['HNSW', 'Indexes']
plain invoke (no tool call): str

import os

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

REPO = "albertoirurueta/docs"          # the only repository this tool may write to


@tool
def open_github_issue(title: str, body: str) -> str:
    """Open an issue in the documentation repository. Destructive: creates public content."""
    token = os.environ["DOCS_ISSUES_TOKEN"]   # fine-grained token, Issues: write on REPO only
    # ... POST https://api.github.com/repos/{REPO}/issues with the token ...
    return f"Opened issue in {REPO}: {title}"


agent = create_agent(
    model,
    tools=[open_github_issue],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"open_github_issue": True})],
    checkpointer=InMemorySaver(),
)
cfg = {"configurable": {"thread_id": "t"}}
out = agent.invoke({"messages": [{"role": "user", "content": "report the broken link"}]}, cfg)
print("interrupted:", "__interrupt__" in out, [i.value["action_requests"][0]["name"] for i in out["__interrupt__"]])
out = agent.invoke(Command(resume={"decisions": [{"type": "reject", "message": "not now"}]}), cfg)
print(out["messages"][-2].content)

interrupted: True ['open_github_issue']
User rejected the tool call for `open_github_issue` with reason: not now