Building AI Agents with LangChain: Tool Integration and Multi-Turn Conversations

User Input → Agent (Reasoning) → Tool Selection → Tool Execution → Response

from langchain.agents import create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Initialize LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# Define prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to tools."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# Create agent
agent = create_openai_tools_agent(llm, tools, prompt)

from langchain_core.tools import tool
from typing import Optional

@tool
def search_knowledge_base(query: str) -> str:
    """Search the private knowledge base for information."""
    # Implementation here
    results = vector_store.similarity_search(query, k=3)
    return "\n".join([doc.page_content for doc in results])

@tool
def get_product_info(product_id: str) -> str:
    """Get detailed product information by ID."""
    # Database query here
    product = db.get_product(product_id)
    return f"Product: {product.name}, Price: {product.price}"

# Register tools
tools = [search_knowledge_base, get_product_info]

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# Initialize memory
memory = ConversationBufferMemory(return_messages=True)

# Create conversation chain
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# Use in agent
agent = create_openai_tools_agent(
    llm, tools, prompt,
    memory=memory
)

from langchain.agents import AgentExecutor

# Create agent executor
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,  # Prevent infinite loops
    handle_parsing_errors=True
)

# Conversation with context
def chat(user_input: str, conversation_id: str):
    # Load conversation history
    history = load_conversation_history(conversation_id)
    
    # Run agent
    result = agent_executor.invoke({
        "input": user_input,
        "chat_history": history
    })
    
    # Save to history
    save_conversation_history(conversation_id, user_input, result["output"])
    
    return result["output"]

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    conversation_id: str

class ChatResponse(BaseModel):
    response: str
    conversation_id: str

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(request: ChatRequest):
    try:
        response = chat(request.message, request.conversation_id)
        return ChatResponse(
            response=response,
            conversation_id=request.conversation_id
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        async for event in agent_executor.astream_events(
            {"input": request.message},
            version="v1"
        ):
            if event["event"] == "on_chat_model_stream":
                content = event["data"]["chunk"].content
                if content:
                    yield f"data: {content}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

@app.middleware("http")
async def error_handler(request, call_next):
    try:
        response = await call_next(request)
        return response
    except Exception as e:
        logger.error(f"Agent error: {e}")
        return JSONResponse(
            status_code=500,
            content={"error": "Internal server error"}
        )

from langchain.agents import create_openai_tools_agent
from langchain.tools import Tool

# Define agricultural tools
tools = [
    Tool(
        name="SearchKnowledgeBase",
        func=search_agricultural_knowledge,
        description="Search private agricultural manuals for crop diseases, treatments, and best practices"
    ),
    Tool(
        name="GeneralAgriculturalKnowledge",
        func=get_general_knowledge,
        description="Get general agricultural information when private knowledge base doesn't have the answer"
    )
]

# System prompt emphasizing private data priority
system_prompt = """You are an agricultural expert assistant. 

IMPORTANT: Always search the private knowledge base FIRST. Only use general knowledge if the private search returns no relevant results.

When answering:
1. Cite the source document when possible
2. Be specific about crop names, disease symptoms, and treatments
3. If unsure, say "I recommend consulting a local agricultural expert"
"""