Agent UX

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_ollama import ChatOllama
from langgraph.config import get_stream_writer

from docsassistant.search import search_docs


@tool
def docs_search_verbose(query: str) -> list[dict]:
    """Search this site's documentation for chunks relevant to `query`."""
    writer = get_stream_writer()
    writer({"status": f"Searching the docs for '{query}'..."})       # a custom progress event
    hits = search_docs(query, k=5)
    writer({"status": f"Found {len(hits)} sections; reading the best ones."})
    return hits


agent = create_agent(ChatOllama(model="llama3.1:8b", temperature=0), tools=[docs_search_verbose])

for mode, chunk in agent.stream(
    {"messages": [("user", "How does DocsAssistant chunk AsciiDoc pages?")]},
    stream_mode=["updates", "messages", "custom"],
):
    if mode == "custom":
        print("[progress]", chunk["status"])                         # show as a status line
    elif mode == "updates":
        for step, update in chunk.items():                           # one entry per finished step
            print("[step]", step)
    elif mode == "messages":
        token, metadata = chunk                                      # LLM tokens as they arrive
        if metadata["langgraph_node"] == "model" and token.content:
            print(token.content, end="", flush=True)

interface StreamingDocsAssistant {
    TokenStream chat(String question);
}

StreamingDocsAssistant assistant = AiServices.builder(StreamingDocsAssistant.class)
        .streamingChatModel(streamingChatModel)          // OllamaStreamingChatModel
        .tools(new DocsTools())
        .build();

assistant.chat("How does DocsAssistant chunk AsciiDoc pages?")
        .onToolExecuted(t -> ui.status("Used " + t.request().name()))   // step updates
        .onPartialResponse(ui::appendToken)                           // answer tokens
        .onCompleteResponse(r -> ui.done())
        .onError(ui::showError)
        .start();

@RestController
class DocsStreamController {

    private final ChatClient docsAssistant;

    DocsStreamController(ChatClient docsAssistant) {
        this.docsAssistant = docsAssistant;
    }

    @GetMapping(path = "/ask", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<String> ask(@RequestParam String q) {
        return docsAssistant.prompt().user(q).stream().content();
    }
}

from typing import Literal

from langchain.agents import create_agent
from langchain_ollama import ChatOllama
from pydantic import BaseModel, Field


class DocsAnswer(BaseModel):
    answer: str
    sources: list[str] = Field(description="page paths the answer is based on")
    confidence: Literal["high", "medium", "low"]
    unsure_about: str | None = Field(None, description="what the sources did not settle, if anything")
    needs_clarification: str | None = Field(None, description="a question to ask the user instead of guessing")


agent = create_agent(
    ChatOllama(model="llama3.1:8b", temperature=0),
    tools=[docs_search],
    response_format=DocsAnswer,
    system_prompt="Answer only from the documentation. If the question is ambiguous, set "
                  "needs_clarification instead of guessing. Rate confidence honestly.",
)

reply = agent.invoke({"messages": [("user", "Which index should I use?")]})["structured_response"]
if reply.needs_clarification:
    ui.ask(reply.needs_clarification)              # ask for guidance rather than guess
else:
    ui.show(reply.answer, sources=reply.sources, badge=reply.confidence, caveat=reply.unsure_about)