Einen ReAct-Agenten mit LangChain erstellen

Das Agent-Ökosystem von LangChain

Einen ReAct-Agenten zusammenbauen

from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 1. Define tools
@tool
def search(query: str) -> str:
    '''Search the web for current information. Input: a search query string.'''
    return f'Search results for "{query}": [mock result]'

@tool
def calculator(expression: str) -> str:
    '''Evaluate a math expression. Input: a valid Python math expression.'''
    try:
        return str(eval(expression))
    except Exception as e:
        return str(e)

tools = [search, calculator]

# 2. Load a ReAct prompt template
prompt = hub.pull('hwchase17/react')

# 3. Create the agent
llm = ChatOpenAI(model='gpt-4o', temperature=0)
agent = create_react_agent(llm, tools, prompt)

Mit AgentExecutor ausführen

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,       # Print each step
    max_iterations=10,  # Safety limit
    handle_parsing_errors=True  # Don't crash on format errors
)

# Invoke with a user question
result = agent_executor.invoke({
    'input': 'What is the square root of 144 multiplied by the number of days in a leap year?'
})

print('Final answer:', result['output'])

Den ausführlichen Trace lesen

# Example verbose output from AgentExecutor:
#
# > Entering new AgentExecutor chain...
# Thought: I need to find the square root of 144 first, then multiply by days in a leap year.
# Action: calculator
# Action Input: 144 ** 0.5
# Observation: 12.0
# Thought: 12.0 times 366 days in a leap year.
# Action: calculator
# Action Input: 12.0 * 366
# Observation: 4392.0
# Thought: I have the final answer.
# Final Answer: 4392.0
# > Finished chain.

Integrierte LangChain-Tools verwenden

from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

# Web search tool
search_tool = DuckDuckGoSearchRun()

# Wikipedia lookup tool
wiki_tool = WikipediaQueryRun(
    api_wrapper=WikipediaAPIWrapper(top_k_results=2)
)

# Add to your agent
tools_with_search = [search_tool, wiki_tool, calculator]
agent_executor.tools = tools_with_search

Dokumentensuche als RAG-Tool

from langchain_core.tools import tool

@tool
def lookup_knowledge_base(query: str) -> str:
    '''Search the company knowledge base for internal policies, procedures, and documentation.
    Use when the user asks about company-specific information not available publicly.
    Input: a natural language question or keyword query.
    '''
    # Your RAG pipeline here
    retrieved_chunks = vector_store.similarity_search(query, k=3)
    context = '\n'.join([doc.page_content for doc in retrieved_chunks])
    return context if context else 'No relevant documents found.'

Agent-Ausgabe streamen

import asyncio

async def stream_agent(user_input: str):
    async for event in agent_executor.astream_events(
        {'input': user_input},
        version='v1'
    ):
        kind = event['event']
        if kind == 'on_chat_model_stream':
            chunk = event['data']['chunk']
            if chunk.content:
                print(chunk.content, end='', flush=True)
        elif kind == 'on_tool_start':
            print(f'\n[Calling tool: {event["name"]}]')
        elif kind == 'on_tool_end':
            print(f'[Tool result: {str(event["data"]["output"])[:100]}]')

asyncio.run(stream_agent('What is the latest news about AI?'))

Agent-Speicher übergeben

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

session_store = {}

def get_session_history(session_id: str):
    if session_id not in session_store:
        session_store[session_id] = ChatMessageHistory()
    return session_store[session_id]

agent_with_memory = RunnableWithMessageHistory(
    agent_executor,
    get_session_history,
    input_messages_key='input',
    history_messages_key='chat_history'
)

Den ReAct-Prompt anpassen

from langchain_core.prompts import ChatPromptTemplate

custom_prompt = ChatPromptTemplate.from_messages([
    ('system', '''You are a helpful customer support agent for TechCorp.
Always be polite and professional. When you do not know something, use your tools.

You have access to these tools:
{tools}

Tool names: {tool_names}

Use this format:
Thought: ...
Action: tool_name
Action Input: input_value
Observation: result
Final Answer: your response to the customer
'''),
    ('human', '{input}'),
    ('assistant', '{agent_scratchpad}')
])

agent = create_react_agent(llm, tools, custom_prompt)

Die Denkschritte des Agenten beobachten

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    return_intermediate_steps=True
)

result = agent_executor.invoke({'input': 'What is 2 to the power of 10?'})

print('Final answer:', result['output'])
print('Reasoning steps:')
for step in result['intermediate_steps']:
    action, observation = step
    print(f'  Tool: {action.tool}, Input: {action.tool_input}')
    print(f'  Observation: {observation}')

Den Agenten gründlich testen

Kurzer Check

Zusammenfassung der Lektion