git clone https://github.com/NirDiamant/agents-towards-production.git
cd agents-towards-production

undefined

undefined

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import os

undefined

from langchain_redis import RedisVectorStore, RedisConfig
from langchain_openai import OpenAIEmbeddings
from langchain.schema import Document
import os

undefined

from contextual_client import ContextualClient
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
import os

context = "\n".join([r.content for r in results])

# Generate response with context
prompt = ChatPromptTemplate.from_template("""
Answer the question based on the following context:

Context: {context}

Question: {question}

Answer:
""")

llm = ChatOpenAI(model="gpt-4", api_key=os.getenv("OPENAI_API_KEY"))
chain = prompt | llm

response = chain.invoke({"context": context, "question": question})
return response.content

undefined

from tavily import TavilyClient
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
import os

# Format results
context = "\n\n".join([
    f"Source: {r['url']}\n{r['content']}"
    for r in search_results.get('results', [])
])

# Generate response
prompt = ChatPromptTemplate.from_template("""
Based on the following web search results, answer the user's question:

Search Results:
{context}

Question: {question}

Provide a comprehensive answer with sources:
""")

llm = ChatOpenAI(model="gpt-4", api_key=os.getenv("OPENAI_API_KEY"))
chain = prompt | llm

response = chain.invoke({"context": context, "question": query})
return {
    "answer": response.content,
    "sources": [r['url'] for r in search_results.get('results', [])]
}

undefined

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, List, Literal
import os

class MultiAgentState(TypedDict):
    messages: List[dict]
    task: str
    current_agent: str
    results: dict

undefined

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import List, Dict
import os
import uvicorn

app = FastAPI(title="Production Agent API")

workflow = StateGraph(dict)
workflow.add_node("process", process)
workflow.set_entry_point("process")
workflow.add_edge("process", END)
return workflow.compile()

    return ChatResponse(
        response=result["messages"][-1]["content"],
        session_id=request.session_id,
        metadata={"user_id": request.user_id}
    )
except Exception as e:
    raise HTTPException(status_code=500, detail=str(e))

undefined

undefined

```yaml

Deploy:

```bash

undefined

from langsmith import Client
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import os
from datetime import datetime

def create_workflow(self):
    """Create observable workflow"""
    def process_with_tracing(state: dict):
        # Start trace
        run_id = langsmith_client.create_run(
            name="agent_process",
            run_type="chain",
            inputs={"messages": state["messages"]},
            project_name=self.project_name
        )
        
        try:
            # Process
            response = self.llm.invoke(state["messages"])
            state["messages"].append({
                "role": "assistant",
                "content": response.content
            })
            
            # Log success
            langsmith_client.update_run(
                run_id=run_id.id,
                outputs={"response": response.content},
                end_time=datetime.now()
            )
        except Exception as e:
            # Log error
            langsmith_client.update_run(
                run_id=run_id.id,
                error=str(e),
                end_time=datetime.now()
            )
            raise
        
        return state
    
    workflow = StateGraph(dict)
    workflow.add_node("process", process_with_tracing)
    workflow.set_entry_point("process")
    workflow.add_edge("process", END)
    return workflow.compile()

undefined

undefined

undefined

from langchain.tools import Tool
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
import os

def calculator(expression: str) -> str:
    """Evaluate mathematical expression"""
    try:
        return str(eval(expression))
    except:
        return "Invalid expression"

def web_search(query: str) -> str:
    """Search the web"""
    # Implementation with Tavily
    return f"Results for: {query}"

tools = [
    Tool(
        name="Calculator",
        func=calculator,
        description="Evaluate mathematical expressions"
    ),
    Tool(
        name="WebSearch",
        func=web_search,
        description="Search the web for information"
    )
]

llm = ChatOpenAI(model="gpt-4", api_key=os.getenv("OPENAI_API_KEY"))
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

result = agent.run("What is 25 * 17? Then search for information about AI agents.")

from langchain_openai import ChatOpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = ChatOpenAI(
    model="gpt-4",
    api_key=os.getenv("OPENAI_API_KEY"),
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

undefined

from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_openai import ChatOpenAI
import os

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def call_llm_with_retry(prompt: str):
    """Call LLM with automatic retries"""
    llm = ChatOpenAI(model="gpt-4", api_key=os.getenv("OPENAI_API_KEY"))
    return llm.invoke([{"role": "user", "content": prompt}])

try:
    response = call_llm_with_retry("Hello, how are you?")
    print(response.content)
except Exception as e:
    print(f"Failed after retries: {e}")

undefined

**Memory not persisting:**
```python

**Rate limiting:**
```python

**Docker deployment issues:**
```bash

**LangGraph state not updating:**
```python

undefined