Observability for LangChain deepagents

from deepagents import create_deep_agent, SubAgent
from langchain_anthropic import ChatAnthropic

from lmnr import Laminar

Laminar.initialize()


def internet_search(query: str) -> str:
    """Search the internet for `query` and return a short snippet."""
    # Replace with a real search (Tavily, Exa, SerpAPI, etc.).
    return f"Top result for '{query}': ..."


research_scout: SubAgent = {
    "name": "research-scout",
    "description": (
        "Researches one or more agent frameworks and returns a concise bulleted "
        "summary covering programming model, strengths, and a typical use case. "
        "Call multiple scouts in parallel when the user asks about several frameworks."
    ),
    "system_prompt": (
        "You are a research-scout. Research the frameworks the parent agent gives "
        "you and return a bulleted summary covering: (1) programming model "
        "(graph, role-play, orchestrator), (2) key strengths, (3) a typical use case."
    ),
    "tools": [internet_search],
}

editor: SubAgent = {
    "name": "editor",
    "description": (
        "Polishes a markdown file in the virtual filesystem. Call once the draft "
        "is written, not during research."
    ),
    "system_prompt": (
        "You are a style editor. Read the markdown file the parent agent gives "
        "you and tighten it under any word limit specified. Preserve all facts; "
        "cut marketing prose. Keep headings and tables intact. The built-in "
        "`read_file` and `edit_file` tools are available from the filesystem "
        "middleware."
    ),
    # Built-in filesystem tools (`read_file`, `edit_file`) are injected by the
    # `FilesystemMiddleware` stack; an empty `tools` list is sufficient.
    "tools": [],
}

agent = create_deep_agent(
    model=ChatAnthropic(model="claude-sonnet-4-5"),
    tools=[internet_search],
    subagents=[research_scout, editor],
    system_prompt=(
        "You are a senior research lead. For comparative questions, dispatch "
        "`research-scout` subagents in parallel via the `task` tool, write the "
        "draft into /final_report.md, then call the `editor` subagent to polish. "
        "Use `write_todos` to plan before you start."
    ),
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": (
            "Write a short briefing comparing LangGraph, CrewAI, and OpenAI Agents "
            "SDK as agent harnesses. For each, cover: programming model "
            "(graph vs. role-play vs. orchestrator), key strengths, and a typical "
            "use case. Use two parallel research-scout subagent calls to gather "
            "material, then synthesize and edit. Keep the final report under 400 "
            "words."
        ),
    }]
})
print(result["messages"][-1].content)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research deepagents and summarize."}]},
    stream_mode="values",
):
    # Each `chunk` is a state snapshot. Print the newest message as it arrives.
    messages = chunk.get("messages", [])
    if messages:
        print(messages[-1].content[:200])

import asyncio

async def main():
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Research deepagents and summarize."}]}
    )
    print(result["messages"][-1].content)

asyncio.run(main())

def fetch_ticket(ticket_id: str) -> dict:
    """Fetch a support ticket by id from the internal database."""
    return {"id": ticket_id, "status": "open", "title": "Timeout on /v1/run"}

agent = create_deep_agent(
    model=ChatAnthropic(model="claude-sonnet-4-5"),
    tools=[fetch_ticket],
    system_prompt="You are a support triage agent. Read the ticket and suggest next steps.",
)